testsocket.cpp 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. ############################################################################## */
  13. #include <platform.h>
  14. #include <stdio.h>
  15. #include "jmisc.hpp"
  16. #include "jlib.hpp"
  17. #include "jsocket.hpp"
  18. #include "jstream.ipp"
  19. #include "portlist.h"
  20. #include "jdebug.hpp"
  21. #include "jthread.hpp"
  22. #include "jfile.hpp"
  23. bool abortEarly = false;
  24. bool forceHTTP = false;
  25. bool abortAfterFirst = false;
  26. bool echoResults = false;
  27. bool saveResults = true;
  28. bool showTiming = false;
  29. bool showStatus = true;
  30. bool sendToSocket = false;
  31. bool parallelBlocked = false;
  32. bool justResults = false;
  33. bool multiThread = false;
  34. bool manyResults = false;
  35. bool sendFileAfterQuery = false;
  36. bool doLock = false;
  37. bool roxieLogMode = false;
  38. StringBuffer sendFileName;
  39. StringAttr queryNameOverride;
  40. unsigned delay = 0;
  41. unsigned runningQueries;
  42. unsigned multiThreadMax;
  43. unsigned maxLineSize = 10000000;
  44. ISocket *persistSocket;
  45. bool persistConnections;
  46. int repeats = 0;
  47. StringBuffer queryPrefix;
  48. Semaphore okToSend;
  49. Semaphore done;
  50. Semaphore finishedReading;
  51. FILE * trace;
  52. CriticalSection traceCrit;
  53. //---------------------------------------------------------------------------
  54. void SplitIpPort(StringAttr & ip, unsigned & port, const char * address)
  55. {
  56. const char * colon = strchr(address, ':');
  57. if (colon)
  58. {
  59. ip.set(address,colon-address);
  60. port = atoi(colon+1);
  61. }
  62. else
  63. ip.set(address);
  64. }
  65. void showMessage(const char * text)
  66. {
  67. if (!justResults)
  68. {
  69. if (echoResults)
  70. fwrite(text, strlen(text), 1, stdout);
  71. if (saveResults && trace != NULL)
  72. fwrite(text, strlen(text), 1, trace);
  73. }
  74. }
  75. void sendFile(const char * filename, ISocket * socket)
  76. {
  77. FILE *in = fopen(filename, "rb");
  78. unsigned size = 0;
  79. void * buff = NULL;
  80. if (in)
  81. {
  82. fseek(in, 0, SEEK_END);
  83. size = ftell(in);
  84. fseek(in, 0, SEEK_SET);
  85. buff = malloc(size);
  86. size_t numRead = fread(buff, 1, size, in);
  87. fclose(in);
  88. if (numRead != size)
  89. {
  90. printf("read from file %s failed (%u/%u)\n", filename, (unsigned)numRead, size);
  91. size = 0;
  92. }
  93. }
  94. else
  95. printf("read from file %s failed\n", filename);
  96. unsigned dllLen = size;
  97. _WINREV(dllLen);
  98. socket->write(&dllLen, sizeof(dllLen));
  99. socket->write(buff, size);
  100. free(buff);
  101. }
  102. #define CHUNK_SIZE 152*2000
  103. void sendFileChunk(const char * filename, offset_t offset, ISocket * socket)
  104. {
  105. FILE *in = fopen(filename, "rb");
  106. unsigned size = 0;
  107. void * buff = NULL;
  108. if (in)
  109. {
  110. fseek(in, 0, SEEK_END);
  111. size = ftell(in);
  112. fseek(in, offset, SEEK_SET);
  113. if (size < offset)
  114. size = 0;
  115. else
  116. size -= offset;
  117. if (size > CHUNK_SIZE)
  118. size = CHUNK_SIZE;
  119. buff = malloc(size);
  120. size_t numRead = fread(buff, 1, size, in);
  121. fclose(in);
  122. if (numRead != size)
  123. {
  124. printf("read from file %s failed (%u/%u)\n", filename, (unsigned)numRead, size);
  125. size = 0;
  126. }
  127. }
  128. else
  129. printf("read from file %s failed\n", filename);
  130. if (size > 0)
  131. {
  132. MemoryBuffer sendBuffer;
  133. unsigned rev = size + strlen(filename) + 10;
  134. rev |= 0x80000000;
  135. _WINREV(rev);
  136. sendBuffer.append(rev);
  137. sendBuffer.append('R');
  138. rev = 0; // should put the sequence number here
  139. _WINREV(rev);
  140. sendBuffer.append(rev);
  141. rev = 0; // should put the # of recs in msg here
  142. _WINREV(rev);
  143. sendBuffer.append(rev);
  144. sendBuffer.append(strlen(filename)+1, filename);
  145. sendBuffer.append(size, buff);
  146. socket->write(sendBuffer.toByteArray(), sendBuffer.length());
  147. }
  148. else
  149. {
  150. unsigned zeroLen = 0;
  151. socket->write(&zeroLen, sizeof(zeroLen));
  152. }
  153. free(buff);
  154. }
  155. int readResults(ISocket * socket, bool readBlocked, bool useHTTP, StringBuffer &result)
  156. {
  157. if (readBlocked)
  158. socket->set_block_mode(BF_SYNC_TRANSFER_PULL,0,60*1000);
  159. unsigned len;
  160. bool is_status;
  161. bool isBlockedResult;
  162. for (;;)
  163. {
  164. if (delay)
  165. MilliSleep(delay);
  166. is_status = false;
  167. isBlockedResult = false;
  168. try
  169. {
  170. if (useHTTP)
  171. len = 0x10000;
  172. else if (readBlocked)
  173. len = socket->receive_block_size();
  174. else
  175. {
  176. socket->read(&len, sizeof(len));
  177. _WINREV(len);
  178. }
  179. }
  180. catch(IException * e)
  181. {
  182. if (manyResults)
  183. showMessage("End of result multiple set\n");
  184. else
  185. pexception("failed to read len data", e);
  186. e->Release();
  187. return 1;
  188. }
  189. if (len == 0)
  190. {
  191. if (manyResults)
  192. {
  193. showMessage("----End of result set----\n");
  194. continue;
  195. }
  196. break;
  197. }
  198. bool isSpecial = false;
  199. bool pluginRequest = false;
  200. bool dataBlockRequest = false;
  201. if (len & 0x80000000)
  202. {
  203. unsigned char flag;
  204. isSpecial = true;
  205. socket->read(&flag, sizeof(flag));
  206. switch (flag)
  207. {
  208. case '-':
  209. if (echoResults)
  210. fputs("Error:", stdout);
  211. if (saveResults && trace != NULL)
  212. fputs("Error:", trace);
  213. break;
  214. case 'D':
  215. showMessage("request for datablock\n");
  216. dataBlockRequest = true;
  217. break;
  218. case 'P':
  219. showMessage("request for plugin\n");
  220. pluginRequest = true;
  221. break;
  222. case 'S':
  223. if (showStatus)
  224. showMessage("Status:");
  225. is_status=true;
  226. break;
  227. case 'T':
  228. showMessage("Timing:\n");
  229. break;
  230. case 'X':
  231. showMessage("---Compound query finished---\n");
  232. return 1;
  233. case 'R':
  234. isBlockedResult = true;
  235. break;
  236. }
  237. len &= 0x7FFFFFFF;
  238. len--; // flag already read
  239. }
  240. char * mem = (char*) malloc(len+1);
  241. char * t = mem;
  242. unsigned sendlen = len;
  243. t[len]=0;
  244. try
  245. {
  246. if (useHTTP)
  247. {
  248. try
  249. {
  250. socket->read(t, 0, len, sendlen);
  251. }
  252. catch (IException *E)
  253. {
  254. if (E->errorCode()!= JSOCKERR_graceful_close)
  255. throw;
  256. E->Release();
  257. break;
  258. }
  259. if (!sendlen)
  260. break;
  261. }
  262. else if (readBlocked)
  263. socket->receive_block(t, len);
  264. else
  265. socket->read(t, len);
  266. }
  267. catch(IException * e)
  268. {
  269. pexception("failed to read data", e);
  270. return 1;
  271. }
  272. if (pluginRequest)
  273. {
  274. //Not very robust! A poor man's implementation for testing...
  275. StringBuffer dllname, libname;
  276. const char * dot = strchr(t, '.');
  277. dllname.append("\\edata\\bin\\debug\\").append(t);
  278. libname.append("\\edata\\bin\\debug\\").append(dot-t,t).append(".lib");
  279. sendFile(dllname.str(), socket);
  280. sendFile(libname.str(), socket);
  281. }
  282. else if (dataBlockRequest)
  283. {
  284. //Not very robust! A poor man's implementation for testing...
  285. offset_t offset;
  286. memcpy(&offset, t, sizeof(offset));
  287. _WINREV(offset);
  288. sendFileChunk(t+sizeof(offset), offset, socket);
  289. }
  290. else
  291. {
  292. if (isBlockedResult)
  293. {
  294. t += 8;
  295. t += strlen(t)+1;
  296. sendlen -= (t - mem);
  297. }
  298. if (echoResults && (!is_status || showStatus))
  299. {
  300. fwrite(t, sendlen, 1, stdout);
  301. fflush(stdout);
  302. }
  303. if (!is_status)
  304. result.append(sendlen, t);
  305. }
  306. free(mem);
  307. if (abortAfterFirst)
  308. return 0;
  309. }
  310. return 0;
  311. }
  312. class ReceiveThread : public Thread
  313. {
  314. public:
  315. virtual int run();
  316. };
  317. int ReceiveThread::run()
  318. {
  319. ISocket * socket = ISocket::create(3456);
  320. ISocket * client = socket->accept();
  321. StringBuffer result;
  322. readResults(client, parallelBlocked, false, result);
  323. client->Release();
  324. socket->Release();
  325. finishedReading.signal();
  326. return 0;
  327. }
  328. //------------------------------------------------------------------------
  329. /**
  330. * Return: 0 - success
  331. * nonzero - error
  332. */
  333. int doSendQuery(const char * ip, unsigned port, const char * base)
  334. {
  335. ISocket * socket;
  336. __int64 starttime, endtime;
  337. StringBuffer ipstr;
  338. try
  339. {
  340. if (strcmp(ip, ".")==0)
  341. ip = GetCachedHostName();
  342. else
  343. {
  344. const char *dash = strchr(ip, '-');
  345. if (dash && isdigit(dash[1]) && dash>ip && isdigit(dash[-1]))
  346. {
  347. if (persistConnections)
  348. UNIMPLEMENTED;
  349. const char *startrange = dash-1;
  350. while (isdigit(startrange[-1]))
  351. startrange--;
  352. char *endptr;
  353. unsigned firstnum = atoi(startrange);
  354. unsigned lastnum = strtol(dash+1, &endptr, 10);
  355. if (lastnum > firstnum)
  356. {
  357. static unsigned counter;
  358. static CriticalSection counterCrit;
  359. CriticalBlock b(counterCrit);
  360. ipstr.append(startrange - ip, ip).append((counter++ % (lastnum+1-firstnum)) + firstnum).append(endptr);
  361. ip = ipstr.str();
  362. printf("Sending to %s\n", ip);
  363. }
  364. }
  365. }
  366. starttime= get_cycles_now();
  367. if (persistConnections)
  368. {
  369. if (!persistSocket) {
  370. SocketEndpoint ep(ip,port);
  371. persistSocket = ISocket::connect_timeout(ep, 1000);
  372. }
  373. socket = persistSocket;
  374. }
  375. else {
  376. SocketEndpoint ep(ip,port);
  377. socket = ISocket::connect_timeout(ep,1000);
  378. }
  379. }
  380. catch(IException * e)
  381. {
  382. pexception("failed to connect to server", e);
  383. return 1;
  384. }
  385. StringBuffer fullQuery;
  386. bool useHTTP = forceHTTP || strstr(base, "<soap:Envelope") != NULL;
  387. if (useHTTP)
  388. {
  389. StringBuffer newQuery;
  390. Owned<IPTree> p = createPTreeFromXMLString(base, ipt_none, ptr_none);
  391. const char *queryName = p->queryName();
  392. if ((stricmp(queryName, "envelope") != 0) && (stricmp(queryName, "envelope") != 0))
  393. {
  394. if (queryNameOverride.length())
  395. queryName = queryNameOverride;
  396. newQuery.appendf("<Envelope xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\"><Body><%sRequest>", queryName);
  397. Owned<IPTreeIterator> elements = p->getElements("./*");
  398. ForEach(*elements)
  399. {
  400. IPTree &elem = elements->query();
  401. toXML(&elem, newQuery, 0, XML_SingleQuoteAttributeValues);
  402. }
  403. newQuery.appendf("</%sRequest></Body></Envelope>", queryName);
  404. base = newQuery.str();
  405. }
  406. // note - don't support queryname override unless original query is xml
  407. fullQuery.appendf("POST /doc HTTP/1.0\r\nContent-Type: application/x-www-form-urlencoded\r\nContent-Length: %d\r\n\r\n", (int) strlen(base)).append(base);
  408. }
  409. else
  410. {
  411. if (sendToSocket)
  412. {
  413. Thread * receive = new ReceiveThread();
  414. receive->start();
  415. receive->Release();
  416. }
  417. if (doLock)
  418. {
  419. const char *lock = "<control:lock/>";
  420. unsigned locklen = strlen(lock);
  421. _WINREV(locklen);
  422. socket->write(&locklen, sizeof(locklen));
  423. socket->write(lock, strlen(lock));
  424. StringBuffer lockResult;
  425. readResults(socket, false, false, lockResult);
  426. }
  427. if (queryNameOverride.length())
  428. {
  429. try
  430. {
  431. Owned<IPTree> p = createPTreeFromXMLString(base, ipt_none, ptr_none);
  432. p->renameProp("/", queryNameOverride);
  433. toXML(p, fullQuery.clear());
  434. }
  435. catch (IException *E)
  436. {
  437. StringBuffer s;
  438. printf("Error: %s", E->errorMessage(s).str());
  439. E->Release();
  440. return 1;
  441. }
  442. }
  443. else
  444. fullQuery.append(base);
  445. }
  446. const char * query = fullQuery.toCharArray();
  447. int len=strlen(query);
  448. int sendlen = len;
  449. if (persistConnections)
  450. sendlen |= 0x80000000;
  451. _WINREV(sendlen);
  452. try
  453. {
  454. if (!useHTTP)
  455. socket->write(&sendlen, sizeof(sendlen));
  456. socket->write(query, len);
  457. if (sendFileAfterQuery)
  458. {
  459. FILE *in = fopen(sendFileName.str(), "rb");
  460. if (in)
  461. {
  462. char buffer[1024];
  463. for (;;)
  464. {
  465. len = fread(buffer, 1, sizeof(buffer), in);
  466. sendlen = len;
  467. _WINREV(sendlen);
  468. socket->write(&sendlen, sizeof(sendlen));
  469. if (!len)
  470. break;
  471. socket->write(buffer, len);
  472. }
  473. fclose(in);
  474. }
  475. else
  476. printf("File %s could not be opened\n", sendFileName.str());
  477. }
  478. }
  479. catch(IException * e)
  480. {
  481. pexception("failed to write data", e);
  482. return 1;
  483. }
  484. if (abortEarly)
  485. return 0;
  486. // back-end does some processing.....
  487. StringBuffer result;
  488. int ret = readResults(socket, false, useHTTP, result);
  489. if ((ret == 0) && !justResults)
  490. {
  491. endtime = get_cycles_now();
  492. CriticalBlock b(traceCrit);
  493. if (trace != NULL)
  494. {
  495. fprintf(trace, "query: %s\n", query);
  496. if (saveResults)
  497. fprintf(trace, "result: %s\n", result.str());
  498. if (showTiming)
  499. fprintf(trace, "Time taken = %.3f msecs\n", (double)(cycle_to_nanosec(endtime - starttime)/1000000));
  500. fputs("----------------------------------------------------------------------------\n", trace);
  501. }
  502. }
  503. if (!persistConnections)
  504. {
  505. socket->close();
  506. socket->Release();
  507. }
  508. return 0;
  509. }
  510. class QueryThread : public Thread
  511. {
  512. public:
  513. QueryThread(const char * _ip, unsigned _port, const char * _base) : ip(_ip),port(_port),base(_base) {}
  514. virtual int run() { doSendQuery(ip, port, base); done.signal(); okToSend.signal(); return 0; }
  515. protected:
  516. StringAttr ip;
  517. unsigned port;
  518. StringAttr base;
  519. };
  520. int sendQuery(const char * ip, unsigned port, const char * base)
  521. {
  522. if (!multiThread)
  523. return doSendQuery(ip, port, base);
  524. if (multiThreadMax)
  525. okToSend.wait();
  526. runningQueries++;
  527. Thread * thread = new QueryThread(ip, port, base);
  528. thread->start();
  529. thread->Release();
  530. return 0;
  531. }
  532. void usage(int exitCode)
  533. {
  534. printf("testsocket ip<:port> [flags] [query | -f[f] file.sql | -]\n");
  535. printf(" - take query from stdin\n");
  536. printf(" -a abort before input recieved\n");
  537. printf(" -a1 abort after first packet receieved\n");
  538. printf(" -c test sending response to a socket\n");
  539. printf(" -cb test sending response to a block mode socket\n");
  540. printf(" -d force delay after each packet\n");
  541. printf(" -f take query from file\n");
  542. printf(" -ff take multiple queries from file, one per line\n");
  543. printf(" -tff take multiple queries from file, one per line, preceded by the time at which it should be submitted (relative to time on first line)\n");
  544. printf(" -k don't save the results to result.txt\n");
  545. printf(" -m only save results to result.txt\n");
  546. printf(" -maxLineSize <n> set maximum query line length\n");
  547. printf(" -n multiple results - keep going until socket closes\n");
  548. printf(" -o set output filename\n");
  549. printf(" -persist use persistant connection\n");
  550. printf(" -pr <text>add a prefix to the query\n");
  551. printf(" -q quiet - don't echo query\n");
  552. printf(" -qname xx Use xx as queryname in place of the xml root element name\n");
  553. printf(" -r <n> repeat the query several times\n");
  554. printf(" -rl roxie logfile mode\n");
  555. printf(" -s add stars to indicate transfer packets\n");
  556. printf(" -ss suppress XML Status messages to screen (always suppressed from tracefile)\n");
  557. printf(" -td add debug timing statistics to trace\n");
  558. printf(" -tf add full timing statistics to trace\n");
  559. printf(" -time add timing to trace\n");
  560. printf(" -u<max> run queries on separate threads\n");
  561. printf(" -cascade cascade query (to all roxie nodes)\n");
  562. printf(" -lock locked cascade query (to all roxie nodes)\n");
  563. exit(exitCode);
  564. }
  565. int main(int argc, char **argv)
  566. {
  567. #ifndef _WIN32
  568. InitModuleObjects();
  569. #endif
  570. StringAttr outputName("result.txt");
  571. bool fromFile = false;
  572. bool fromStdIn = false;
  573. bool fromMultiFile = false;
  574. bool timedReplay = false;
  575. if (argc < 2 && !(argc==2 && strstr(argv[1], "::")))
  576. usage(1);
  577. int arg = 2;
  578. bool echoSingle = true;
  579. while (arg < argc && *argv[arg]=='-')
  580. {
  581. if (stricmp(argv[arg], "-time") == 0)
  582. {
  583. showTiming = true;
  584. ++arg;
  585. }
  586. else if (stricmp(argv[arg], "-a") == 0)
  587. {
  588. abortEarly = true;
  589. ++arg;
  590. }
  591. else if (stricmp(argv[arg], "-a1") == 0)
  592. {
  593. abortAfterFirst = true;
  594. ++arg;
  595. }
  596. else if (stricmp(argv[arg], "-c") == 0)
  597. {
  598. sendToSocket = true;
  599. ++arg;
  600. }
  601. else if (stricmp(argv[arg], "-cb") == 0)
  602. {
  603. sendToSocket = true;
  604. parallelBlocked = true;
  605. ++arg;
  606. }
  607. else if (stricmp(argv[arg], "-d") == 0)
  608. {
  609. delay = 300;
  610. ++arg;
  611. }
  612. else if (stricmp(argv[arg], "-http") == 0)
  613. {
  614. forceHTTP = true;
  615. ++arg;
  616. }
  617. else if (stricmp(argv[arg], "-") == 0)
  618. {
  619. fromStdIn = true;
  620. ++arg;
  621. }
  622. else if (stricmp(argv[arg], "-f") == 0)
  623. {
  624. fromFile = true;
  625. ++arg;
  626. }
  627. else if (stricmp(argv[arg], "-ff") == 0)
  628. {
  629. fromMultiFile = true;
  630. ++arg;
  631. }
  632. else if (stricmp(argv[arg], "-tff") == 0)
  633. {
  634. fromMultiFile = true;
  635. timedReplay = true;
  636. ++arg;
  637. }
  638. else if (stricmp(argv[arg], "-k") == 0)
  639. {
  640. saveResults = false;
  641. ++arg;
  642. }
  643. else if (stricmp(argv[arg], "-m") == 0)
  644. {
  645. justResults = true;
  646. ++arg;
  647. }
  648. else if (stricmp(argv[arg], "-maxlinesize") == 0)
  649. {
  650. ++arg;
  651. if (arg>=argc)
  652. usage(1);
  653. maxLineSize = atoi(argv[arg]);
  654. ++arg;
  655. }
  656. else if (stricmp(argv[arg], "-n") == 0)
  657. {
  658. manyResults = true;
  659. ++arg;
  660. }
  661. else if (stricmp(argv[arg], "-o") == 0)
  662. {
  663. outputName.set(argv[arg+1]);
  664. arg+=2;
  665. }
  666. else if (stricmp(argv[arg], "-persist") == 0)
  667. {
  668. persistConnections = true;
  669. ++arg;
  670. }
  671. else if (stricmp(argv[arg], "-pr") == 0)
  672. {
  673. queryPrefix.append(argv[arg+1]);
  674. arg+=2;
  675. }
  676. else if (stricmp(argv[arg], "-q") == 0)
  677. {
  678. echoSingle = false;
  679. ++arg;
  680. }
  681. else if (stricmp(argv[arg], "-qname") == 0)
  682. {
  683. queryNameOverride.set(argv[arg+1]);
  684. arg+=2;
  685. }
  686. else if (stricmp(argv[arg], "-r") == 0)
  687. {
  688. ++arg;
  689. if (arg>=argc)
  690. usage(1);
  691. repeats = atoi(argv[arg]);
  692. ++arg;
  693. }
  694. else if (stricmp(argv[arg], "-rl") == 0)
  695. {
  696. roxieLogMode = true;
  697. fromMultiFile = true;
  698. ++arg;
  699. }
  700. else if (stricmp(argv[arg], "-ss") == 0)
  701. {
  702. showStatus = false;
  703. ++arg;
  704. }
  705. else if (memicmp(argv[arg], "-u", 2) == 0)
  706. {
  707. multiThread = true;
  708. multiThreadMax = atoi(argv[arg]+2);
  709. if (multiThreadMax)
  710. okToSend.signal(multiThreadMax);
  711. ++arg;
  712. }
  713. else if (stricmp(argv[arg], "-lock") == 0)
  714. {
  715. doLock = true;
  716. ++arg;
  717. }
  718. else if (memicmp(argv[arg], "-z", 2) == 0)
  719. {
  720. sendFileAfterQuery = true;
  721. sendFileName.append(argv[arg+1]);
  722. OwnedIFile f = createIFile(sendFileName.str());
  723. if (!f->exists() || !f->isFile())
  724. {
  725. printf("file %s does not exist\n", sendFileName.str());
  726. exit (EXIT_FAILURE);
  727. }
  728. arg+=2;
  729. }
  730. else
  731. {
  732. printf("Unknown argument %s, ignored\n", argv[arg]);
  733. ++arg;
  734. }
  735. }
  736. if (persistConnections && multiThread)
  737. {
  738. printf("Multi-thread (-u) not available with -persist - ignored\n");
  739. multiThread = false;
  740. }
  741. StringAttr ip;
  742. unsigned socketPort = ROXIE_SERVER_PORT;
  743. SplitIpPort(ip, socketPort, argv[1]);
  744. int ret = 0;
  745. trace = fopen(outputName, "w");
  746. if (trace == NULL)
  747. {
  748. printf("Can't open %s for writing\n", outputName.str());
  749. }
  750. __int64 starttime,endtime;
  751. starttime = get_cycles_now();
  752. if (arg < argc || fromStdIn)
  753. {
  754. echoResults = echoSingle;
  755. do
  756. {
  757. const char * query = argv[arg];
  758. if (fromMultiFile)
  759. {
  760. FILE *in = fopen(query, "rt");
  761. if (in)
  762. {
  763. CDateTime firstTime;
  764. CDateTime startTime;
  765. bool firstLine = true;
  766. char *buffer = new char[maxLineSize];
  767. for (;;)
  768. {
  769. if (fgets(buffer, maxLineSize, in)==NULL) // buffer overflow possible - do I care?
  770. break;
  771. if (timedReplay)
  772. {
  773. if (firstLine)
  774. {
  775. firstTime.setNow();
  776. firstTime.setTimeString(buffer, &query);
  777. startTime.setNow();
  778. }
  779. else
  780. {
  781. CDateTime queryTime, nowTime;
  782. queryTime.setNow();
  783. queryTime.setTimeString(buffer, &query);
  784. nowTime.setNow();
  785. int sleeptime = (queryTime.getSimple()-firstTime.getSimple()) - (nowTime.getSimple()-startTime.getSimple());
  786. if (sleeptime < 0)
  787. DBGLOG("Running behind %d seconds", -sleeptime);
  788. else if (sleeptime)
  789. {
  790. DBGLOG("Sleeping %d seconds", sleeptime);
  791. Sleep(sleeptime*1000);
  792. }
  793. StringBuffer targetTime;
  794. queryTime.getTimeString(targetTime);
  795. DBGLOG("Virtual time is %s", targetTime.str());
  796. }
  797. }
  798. else
  799. {
  800. query = buffer;
  801. while (isspace(*query)) query++;
  802. if (roxieLogMode)
  803. {
  804. char *start = (char *) strchr(query, '<');
  805. if (start)
  806. {
  807. char *end = (char *) strchr(start, '"');
  808. if (end && end[1]=='\n')
  809. {
  810. query = start;
  811. *end = 0;
  812. }
  813. }
  814. }
  815. }
  816. if (query)
  817. {
  818. ret = sendQuery(ip, socketPort, query);
  819. firstLine = false;
  820. }
  821. }
  822. delete [] buffer;
  823. fclose(in);
  824. }
  825. else
  826. printf("File %s could not be opened\n", query);
  827. }
  828. else if (fromFile || fromStdIn)
  829. {
  830. FILE *in = fromStdIn ? stdin : fopen(query, "rt");
  831. if (in)
  832. {
  833. StringBuffer fileContents;
  834. char buffer[1024];
  835. int bytes;
  836. for (;;)
  837. {
  838. bytes = fread(buffer, 1, sizeof(buffer), in);
  839. if (!bytes)
  840. break;
  841. fileContents.append(buffer, 0, bytes);
  842. }
  843. if (in != stdin)
  844. fclose(in);
  845. ret = sendQuery(ip, socketPort, fileContents.toCharArray());
  846. }
  847. else
  848. printf("File %s could not be opened\n", query);
  849. }
  850. else
  851. {
  852. ret = sendQuery(ip, socketPort, query);
  853. if (sendToSocket)
  854. finishedReading.wait();
  855. }
  856. } while (--repeats > 0);
  857. }
  858. else
  859. usage(2);
  860. while (runningQueries--)
  861. done.wait();
  862. if (persistConnections && persistSocket)
  863. {
  864. int sendlen=0;
  865. persistSocket->write(&sendlen, sizeof(sendlen));
  866. persistSocket->close();
  867. persistSocket->Release();
  868. }
  869. endtime = get_cycles_now();
  870. if (!justResults)
  871. {
  872. if (trace != NULL)
  873. {
  874. fprintf(trace, "Total Time taken = %.3f msecs\n", (double)(cycle_to_nanosec(endtime - starttime)/1000000));
  875. fputs("----------------------------------------------------------------------------\n", trace);
  876. }
  877. }
  878. if (trace != NULL)
  879. {
  880. fclose(trace);
  881. }
  882. #ifdef _DEBUG
  883. releaseAtoms();
  884. #endif
  885. return ret;
  886. }