hce-node application  1.4.3
HCE Hierarchical Cluster Engine node application
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros
SphinxMultiIndexTest.cpp
Go to the documentation of this file.
1 #include <Poco/SharedPtr.h>
2 #include <Poco/ScopedLock.h>
3 #include <Poco/ThreadPool.h>
4 #include <fstream>
5 #include <assert.h>
6 #include <iomanip>
7 #include <map>
8 
9 #include "SphinxMultiIndexTest.hpp"
10 #include "SphinxManageTest.hpp"
11 #include "SphinxSearchTest.hpp"
12 #include "SphinxIndexTest.hpp"
13 #include "SphinxAdminCommand.hpp"
14 #include "SphinxInputJsonMessage.hpp"
15 #include "SphinxOutputJsonMessage.hpp"
16 #include "SphinxConfigCreator.hpp"
17 #include "SphinxInputJsonMessageSearch.hpp"
18 #include "SphinxPrintStatus.hpp"
19 
20 namespace HCE
21 {
22 namespace sphinx
23 {
24 namespace tests
25 {
26 //-----------------------------------------------------------------------------
27 SearchTask::SearchTask(const std::string& indexName_, std::vector<std::string>& queries_, bool start, bool stop)
28 : inherited("search_"+indexName_), indexName(indexName_), queries(queries_),
29  fObj("node_"+indexName_, sphinx_search_const::defaultDataDir, indexName_, start, stop), fps(0)
30 {
31  if (!start)
32  fObj.easyStart();
33 }
34 //-----------------------------------------------------------------------------
35 // cppcheck-suppress unusedFunction
36 void SearchTask::runTask(void)
37 {
38  fps = 0;
39  if (queries.size())
40  {
41  float localTime = 0;
42  float progress = 0;
43  const float progressStep = 100/queries.size();
44 
45  for (size_t i=0;i<queries.size();++i)
46  {
47  localTime += SphinxMultiIndexTest::testSearchIndex(fObj, indexName, queries[i]);
48  progress += progressStep;
49  setProgress(progress);
50  }
51 
52  localTime /= 1000;
53  localTime /= queries.size();
54  fps = 1/localTime;
55  }
56 }
57 //-----------------------------------------------------------------------------
58 //-----------------------------------------------------------------------------
59 void SearchProgressHandler::onProgress(Poco::TaskProgressNotification* pNf)
60 {
61  std::cout << pNf->task()->name() << " progress: " << pNf->progress() << std::endl;
62  pNf->release();
63 }
64 //-----------------------------------------------------------------------------
65 void SearchProgressHandler::onFinished(Poco::TaskFinishedNotification* pNf)
66 {
67  std::cout << pNf->task()->name() << " finished." << std::endl;
68 
69  Poco::Mutex::ScopedLock lock(mutex);
70  fps += static_cast<SearchTask*>(pNf->task())->getFps();
71 
72  pNf->release();
73 }
74 //-----------------------------------------------------------------------------
75 float SearchProgressHandler::getFps(void)
76 {
77  Poco::Mutex::ScopedLock lock(mutex);
78  return fps;
79 }
80 //-----------------------------------------------------------------------------
81 //-----------------------------------------------------------------------------
82 void SphinxMultiIndexTest::printSuccess(const std::string& msg, std::ostream& os)
83 {
84  os << std::setw(40) << msg << " - OK\n";
85 }
86 //-----------------------------------------------------------------------------
87 void SphinxMultiIndexTest::printWarning(const std::string& msg, std::ostream& os)
88 {
89  os << std::setw(40) << msg << " - WARNING\n";
90 }
91 //-----------------------------------------------------------------------------
92 //-----------------------------------------------------------------------------
93 void SphinxMultiIndexTest::testCreateIndex(SphinxFunctionalObject& fObj, const std::string& indexName, const std::string& dataContent)
94 {
95  SphinxManageTest::testIndexCreate(fObj, indexName);
96  SphinxManageTest::testIndexCheck(fObj, indexName, true);
97 
98  std::vector<std::string> branches;
99  branches.push_back("branch_"+indexName);
100 
101  if (!dataContent.empty())
102  {
103  for (size_t i=0;i<branches.size();++i)
104  SphinxManageTest::testIndexStoreDataFile(fObj, indexName, branches[i], dataContent);// add data content
105  SphinxMultiIndexTest::printSuccess("Using data file content", fObj.log(HCE::LoggerStream::PRIO_INFORMATION));
106  }
107  else
108  {
109  for (size_t i=0;i<branches.size();++i)
110  SphinxManageTest::testIndexStoreDataFile(fObj, indexName, branches[i]);
111  SphinxMultiIndexTest::printWarning("Using internal test data content", fObj.log(HCE::LoggerStream::PRIO_INFORMATION));
112  }
113 
114  SphinxManageTest::testIndexStoreSchemaFile(fObj, indexName);
115  SphinxManageTest::testIndexRebuild(fObj, indexName, branches);
116  SphinxManageTest::testIndexMerge(fObj, indexName, branches);
117 }
118 //-----------------------------------------------------------------------------
119 unsigned int SphinxMultiIndexTest::testSearchIndex(SphinxFunctionalObject& fObj, const std::string& indexName, const std::string& query)
120 {
121  std::string json;
122  SphinxInputJsonMessageSearch msgSearch;
123  msgSearch.setQueryString(query);
124  // msgSearch.setMaxResultsNumber(5);
125  msgSearch.addQueryParameters("queryId", "777");
126  msgSearch.addQueryParameters("JsonType", "7");
127  msgSearch.addQueryParameters("offset", "0");
128  msgSearch.addQueryParameters("limit", "0");
129  msgSearch.addQueryParameters("cutoff", "0");
130  assert(msgSearch.serialize(json));
131  printSuccess("Serialize SphinxInputJsonMessageSearch", fObj.log(HCE::LoggerStream::PRIO_INFORMATION));
132 
133  SphinxInputJsonMessage handler;
134  handler.setType(SphinxInputJsonMessage::MessageType::mtSearch);
135  handler.setData(json);
136 
137  std::stringstream str;
138  str << handler;
139 
140  std::string ret = fObj.Process(str.str());
141 
142 // std::cout << "ret: " << ret << std::endl;
143  if (!fObj.getErrorMsg().empty())
144  std::cout << "fObj.getErrorMsg() = " << fObj.getErrorMsg() << std::endl;
145 
146  if (!fObj.getErrorMsg().empty())
147  fObj.log(HCE::LoggerStream::PRIO_INFORMATION) << "fObj.getErrorMsg() = " << fObj.getErrorMsg() << std::endl;
148  assert(fObj.getErrorMsg().empty());
149 
150  SphinxOutputJsonMessage outputMessage(ret);
151  assert(!outputMessage.isError());
152 
153  fObj.log(HCE::LoggerStream::PRIO_INFORMATION) << "Time = " << outputMessage.getTime() << std::endl;
154  printSuccess("Search Process Query", fObj.log(HCE::LoggerStream::PRIO_INFORMATION));
155  return outputMessage.getTime();
156 }
157 //-----------------------------------------------------------------------------
158 void SphinxMultiIndexTest::makeIndexNames(size_t count, std::vector<std::string>& names)
159 {
160  names.clear();
161  for (size_t i=1;i<=count;++i)
162  {
163  std::stringstream indexName;
164  indexName << "i";
165  indexName.width(3);
166  indexName.fill('0');
167  indexName << i;
168  names.push_back(indexName.str());
169  }
170 }
171 //-----------------------------------------------------------------------------
172 void SphinxMultiIndexTest::testCreateIndexes(const std::string& homeDir, size_t count, const std::string& dataFileName, std::vector<std::string>& names)
173 {
174  std::string dataContent;
175  std::ifstream ifs(dataFileName.c_str(), std::fstream::binary);
176  if (ifs.is_open())
177  {
178  std::streambuf* pbuf = ifs.rdbuf();
179  std::streamsize size = pbuf->pubseekoff(0,ifs.end);
180  pbuf->pubseekoff(0,ifs.beg); // rewind
181  dataContent.resize(size);
182  pbuf->sgetn (const_cast<char*>(dataContent.c_str()), size);
183  ifs.close();
184  printSuccess("Load data file for test '"+dataFileName+"'");
185  }
186 
187  SphinxFunctionalObject fObj("test_node", homeDir);
188 // fObj.loadConfiguration(sphinx_search_const::defaultIndexName);
189 // fObj.setLogStream(std::cout);
190 // if (!fObj.isLoadedConfig())
191 // std::cout << "load config: '" << fObj.getErrorMsg() << "' errorCode = " << fObj.getErrorCode() << std::endl;
192 // assert(fObj.isLoadedConfig());
193 
194  SphinxMultiIndexTest::makeIndexNames(count, names);
195 
196  for (size_t i=0;i<names.size();++i)
197  {
198  SphinxMultiIndexTest::testCreateIndex(fObj, names[i], dataContent);
199  }
200 }
201 //-----------------------------------------------------------------------------
202 float SphinxMultiIndexTest::testSearchIndexes(std::vector<std::string>& names, std::vector<std::string> queries)
203 {
204  assert(names.size());
205  Poco::ThreadPool threadPool(1, names.size());
206  Poco::TaskManager tm(threadPool);
207  SearchProgressHandler pm;
208  tm.addObserver(Poco::Observer<SearchProgressHandler, Poco::TaskProgressNotification>(pm, &SearchProgressHandler::onProgress));
209  tm.addObserver(Poco::Observer<SearchProgressHandler, Poco::TaskFinishedNotification>(pm, &SearchProgressHandler::onFinished));
210 
211  for (size_t i=0;i<names.size();++i)
212  {
213  tm.start(new SearchTask(names[i], queries));
214  }
215  tm.joinAll();
216 
217  return pm.getFps();
218 }
219 //-----------------------------------------------------------------------------
220 void SphinxMultiIndexTest::setIndexProperty(const std::string& homeDir, const std::string& indexName, const IndexProperty& property) throw (Poco::Exception)
221 {
222  std::string propFile = homeDir+"/"+sphinx_search_const::defaultDataDir+"/"+indexName+"/"+sphinx_search_const::sphinxPropertyFile;
223  Poco::AutoPtr<SphinxConfigCreator> pConf = new SphinxConfigCreator(propFile);
224 
225  std::vector<std::string> keys;
226  pConf->keys(keys);
227 
228  for (size_t i=0;i<keys.size();++i)
229  {
230  if (keys[i].find(sphinx_search_const::indexNameField)!=std::string::npos)
231  {
232  std::string name = keys[i]+"."+sphinx_search_const::pathNameField;
233  pConf->setString(name, property.path);
234  }
235  }
236  pConf->setString(sphinx_search_const::searchdListen, property.listen);
237 
238  pConf->save(propFile);
239  pConf->setPropertyFileConfiguration(propFile);
240 
241  std::string sphinxConfig = sphinx_search_const::defaultDataDir+"/"+indexName+"/"+sphinx_search_const::sphinxConfigName;
242  std::ofstream ofs(sphinxConfig.c_str(), std::fstream::trunc);
243  ofs << *pConf;
244  ofs.close();
245 }
246 //-----------------------------------------------------------------------------
247 void SphinxMultiIndexTest::getIndexProperty(const std::string& homeDir, const std::string& indexName, IndexProperty& property) throw (Poco::Exception)
248 {
249  std::string propFile = homeDir+"/"+sphinx_search_const::defaultDataDir+"/"+indexName+"/"+sphinx_search_const::sphinxPropertyFile;
250  Poco::AutoPtr<SphinxConfigCreator> pConf = new SphinxConfigCreator(propFile);
251 
252  std::vector<std::string> keys;
253  pConf->keys(keys);
254 
255  for (size_t i=0;i<keys.size();++i)
256  {
257  if (keys[i].find(sphinx_search_const::indexNameField)!=std::string::npos)
258  {
259  std::string name = keys[i]+"."+sphinx_search_const::pathNameField;
260  property.path = pConf->getString(name);
261  break;
262  }
263  }
264  property.listen = pConf->getString(sphinx_search_const::searchdListen);
265 }
266 //-----------------------------------------------------------------------------
267 float SphinxMultiIndexTest::testSearchIndexes(const std::string& homeDir, std::vector<std::string>& names, std::vector<std::string> queries, const std::string& indexName)
268 {
269  assert(names.size());
270 
271  Poco::ThreadPool threadPool(1, names.size());
272  Poco::TaskManager tm(threadPool);
273  SearchProgressHandler pm;
274  tm.addObserver(Poco::Observer<SearchProgressHandler, Poco::TaskProgressNotification>(pm, &SearchProgressHandler::onProgress));
275  tm.addObserver(Poco::Observer<SearchProgressHandler, Poco::TaskFinishedNotification>(pm, &SearchProgressHandler::onFinished));
276 
277  try
278  {
279  IndexProperty property;
280  SphinxMultiIndexTest::getIndexProperty(homeDir, indexName, property); // get value for applicable to continue next
281 
282  // correct configuration files
283  std::map<std::string, IndexProperty> Map;
284  for (size_t i=0;i<names.size();++i)
285  {
286  if (indexName==names[i])
287  continue;
288  IndexProperty localProperty;
289  SphinxMultiIndexTest::getIndexProperty(homeDir, names[i], localProperty);
290  Map[names[i]] = localProperty;
291  SphinxMultiIndexTest::setIndexProperty(homeDir, names[i], property);
292  }
293 
294  // next make search
295  tm.start(new SearchTask(indexName, queries, true, false));
296  for (size_t i=0;i<names.size();++i)
297  {
298  if (indexName==names[i])
299  continue;
300  tm.start(new SearchTask(names[i], queries, false, false));
301  }
302  tm.joinAll();
303 
304  SphinxFunctionalObject fObj("node_"+indexName, homeDir, indexName, false, true);
305 
306  // restore configuration files
307  for (std::map<std::string, IndexProperty>::iterator iter=Map.begin();iter!=Map.end();++iter)
308  {
309 // std::cout << "[" << (*iter).first << "] = " << (*iter).second.path << "\t" << (*iter).second.listen << std::endl;
310  SphinxMultiIndexTest::setIndexProperty(homeDir, (*iter).first, (*iter).second);
311  }
312  }
313  catch(Poco::Exception& e)
314  {
315  std::cout << "Error ("<< e.code() << "): " << e.message() << std::endl;
316  }
317  return pm.getFps();
318 }
319 //-----------------------------------------------------------------------------
320 void SphinxMultiIndexTest::testRemoveIndexes(const std::string& homeDir, std::vector<std::string>& names)
321 {
322  SphinxFunctionalObject fObj("test_node", homeDir);//, sphinx_search_const::defaultIndexName);
323  for (size_t i=0;i<names.size();++i)
324  {
325  SphinxManageTest::testIndexRemove(fObj, names[i]);
326  }
327  names.clear();
328 }
329 //-----------------------------------------------------------------------------
330 //-----------------------------------------------------------------------------
331 } // namespace tests
332 } // namespace sphinx
333 } // namespace HCE