testsocket.cpp 28 KB

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