ccdprotocol.cpp 70 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2016 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 "jlib.hpp"
  15. #include "jthread.hpp"
  16. #include "roxie.hpp"
  17. #include "roxiehelper.hpp"
  18. #include "ccdprotocol.hpp"
  19. //================================================================================================================================
  20. IHpccProtocolListener *createProtocolListener(const char *protocol, IHpccProtocolMsgSink *sink, unsigned port, unsigned listenQueue);
  21. class CHpccProtocolPlugin : public CInterface, implements IHpccProtocolPlugin
  22. {
  23. public:
  24. IMPLEMENT_IINTERFACE;
  25. CHpccProtocolPlugin(IHpccProtocolPluginContext &ctx)
  26. {
  27. targetNames.appendListUniq(ctx.ctxQueryProp("@querySets"), ",");
  28. targetAliases.setown(createProperties());
  29. StringArray tempList;
  30. tempList.appendListUniq(ctx.ctxQueryProp("@targetAliases"), ",");
  31. ForEachItemIn(i, tempList)
  32. {
  33. const char *alias = tempList.item(i);
  34. const char *eq = strchr(alias, '=');
  35. if (eq)
  36. {
  37. StringAttr name(alias, eq-alias);
  38. if (!targetNames.contains(name))
  39. targetAliases->setProp(name.str(), ++eq);
  40. }
  41. }
  42. maxBlockSize = ctx.ctxGetPropInt("@maxBlockSize", 10000000);
  43. defaultXmlReadFlags = ctx.ctxGetPropBool("@defaultStripLeadingWhitespace", true) ? ptr_ignoreWhiteSpace : ptr_none;
  44. trapTooManyActiveQueries = ctx.ctxGetPropBool("@trapTooManyActiveQueries", true);
  45. numRequestArrayThreads = ctx.ctxGetPropInt("@requestArrayThreads", 5);
  46. }
  47. IHpccProtocolListener *createListener(const char *protocol, IHpccProtocolMsgSink *sink, unsigned port, unsigned listenQueue, const char *config)
  48. {
  49. return createProtocolListener(protocol, sink, port, listenQueue);
  50. }
  51. public:
  52. StringArray targetNames;
  53. Owned<IProperties> targetAliases;
  54. PTreeReaderOptions defaultXmlReadFlags;
  55. unsigned maxBlockSize;
  56. unsigned numRequestArrayThreads;
  57. bool trapTooManyActiveQueries;
  58. };
  59. Owned<CHpccProtocolPlugin> global;
  60. class ProtocolListener : public Thread, implements IHpccProtocolListener, implements IThreadFactory
  61. {
  62. public:
  63. IMPLEMENT_IINTERFACE;
  64. ProtocolListener(IHpccProtocolMsgSink *_sink) : Thread("RoxieListener")
  65. {
  66. running = false;
  67. sink.setown(dynamic_cast<IHpccNativeProtocolMsgSink*>(_sink));
  68. }
  69. virtual IHpccProtocolMsgSink *queryMsgSink()
  70. {
  71. return sink;
  72. }
  73. static void updateAffinity()
  74. {
  75. #ifdef CPU_ZERO
  76. if (sched_getaffinity(0, sizeof(cpu_set_t), &cpuMask))
  77. {
  78. if (traceLevel)
  79. DBGLOG("Unable to get CPU affinity - thread affinity settings will be ignored");
  80. cpuCores = 0;
  81. lastCore = 0;
  82. CPU_ZERO(&cpuMask);
  83. }
  84. else
  85. {
  86. #if __GLIBC__ > 2 || (__GLIBC__ == 2 && __GLIBC_MINOR__ >= 6)
  87. cpuCores = CPU_COUNT(&cpuMask);
  88. #else
  89. cpuCores = 0;
  90. unsigned setSize = CPU_SETSIZE;
  91. while (setSize--)
  92. {
  93. if (CPU_ISSET(setSize, &cpuMask))
  94. ++cpuCores;
  95. }
  96. #endif /* GLIBC */
  97. if (traceLevel)
  98. traceAffinity(&cpuMask);
  99. }
  100. #endif
  101. }
  102. virtual void start()
  103. {
  104. // Note we allow a few additional threads than requested - these are the threads that return "Too many active queries" responses
  105. pool.setown(createThreadPool("RoxieSocketWorkerPool", this, NULL, sink->getPoolSize()+5, INFINITE));
  106. assertex(!running);
  107. Thread::start();
  108. started.wait();
  109. }
  110. virtual bool stop(unsigned timeout)
  111. {
  112. if (running)
  113. {
  114. running = false;
  115. join();
  116. Release();
  117. }
  118. return pool->joinAll(false, timeout);
  119. }
  120. virtual bool suspend(bool suspendIt)
  121. {
  122. return sink->suspend(suspendIt);
  123. }
  124. void setThreadAffinity(int numCores)
  125. {
  126. #ifdef CPU_ZERO
  127. // Note - strictly speaking not threadsafe but any race conditions are (a) unlikely and (b) harmless
  128. if (cpuCores)
  129. {
  130. if (numCores > 0 && numCores < cpuCores)
  131. {
  132. cpu_set_t threadMask;
  133. CPU_ZERO(&threadMask);
  134. unsigned cores = 0;
  135. unsigned offset = lastCore;
  136. unsigned core;
  137. for (core = 0; core < CPU_SETSIZE; core++)
  138. {
  139. unsigned useCore = (core + offset) % CPU_SETSIZE;
  140. if (CPU_ISSET(useCore, &cpuMask))
  141. {
  142. CPU_SET(useCore, &threadMask);
  143. cores++;
  144. if (cores == numCores)
  145. {
  146. lastCore = useCore+1;
  147. break;
  148. }
  149. }
  150. }
  151. if (traceLevel > 3)
  152. traceAffinity(&threadMask);
  153. pthread_setaffinity_np(GetCurrentThreadId(), sizeof(cpu_set_t), &threadMask);
  154. }
  155. else
  156. {
  157. if (traceLevel > 3)
  158. traceAffinity(&cpuMask);
  159. pthread_setaffinity_np(GetCurrentThreadId(), sizeof(cpu_set_t), &cpuMask);
  160. }
  161. }
  162. #endif
  163. }
  164. protected:
  165. bool running;
  166. Semaphore started;
  167. Owned<IThreadPool> pool;
  168. Linked<IHpccNativeProtocolMsgSink> sink;
  169. #ifdef CPU_ZERO
  170. static cpu_set_t cpuMask;
  171. static unsigned cpuCores;
  172. static unsigned lastCore;
  173. private:
  174. static void traceAffinity(cpu_set_t *mask)
  175. {
  176. StringBuffer trace;
  177. for (unsigned core = 0; core < CPU_SETSIZE; core++)
  178. {
  179. if (CPU_ISSET(core, mask))
  180. trace.appendf(",%d", core);
  181. }
  182. if (trace.length())
  183. DBGLOG("Process affinity is set to use core(s) %s", trace.str()+1);
  184. }
  185. #endif
  186. };
  187. #ifdef CPU_ZERO
  188. cpu_set_t ProtocolListener::cpuMask;
  189. unsigned ProtocolListener::cpuCores;
  190. unsigned ProtocolListener::lastCore;
  191. #endif
  192. class ProtocolSocketListener : public ProtocolListener
  193. {
  194. unsigned port;
  195. unsigned listenQueue;
  196. Owned<ISocket> socket;
  197. SocketEndpoint ep;
  198. public:
  199. ProtocolSocketListener(IHpccProtocolMsgSink *_sink, unsigned _port, unsigned _listenQueue)
  200. : ProtocolListener(_sink)
  201. {
  202. port = _port;
  203. listenQueue = _listenQueue;
  204. ep.set(port, queryHostIP());
  205. }
  206. IHpccProtocolMsgSink *queryMsgSink()
  207. {
  208. return sink.get();
  209. }
  210. virtual bool stop(unsigned timeout)
  211. {
  212. if (socket)
  213. socket->cancel_accept();
  214. return ProtocolListener::stop(timeout);
  215. }
  216. virtual void disconnectQueue()
  217. {
  218. // This is for dali queues only
  219. }
  220. virtual void stopListening()
  221. {
  222. // Not threadsafe, but we only call this when generating a core file... what's the worst that can happen?
  223. try
  224. {
  225. DBGLOG("Closing listening socket %d", port);
  226. socket.clear();
  227. DBGLOG("Closed listening socket %d", port);
  228. }
  229. catch(...)
  230. {
  231. }
  232. }
  233. virtual void runOnce(const char *query);
  234. virtual int run()
  235. {
  236. DBGLOG("ProtocolSocketListener (%d threads) listening to socket on port %d", sink->getPoolSize(), port);
  237. socket.setown(ISocket::create(port, listenQueue));
  238. running = true;
  239. started.signal();
  240. while (running)
  241. {
  242. ISocket *client = socket->accept(true);
  243. if (client)
  244. {
  245. client->set_linger(-1);
  246. pool->start(client);
  247. }
  248. }
  249. DBGLOG("ProtocolSocketListener closed query socket");
  250. return 0;
  251. }
  252. virtual IPooledThread *createNew();
  253. virtual const SocketEndpoint &queryEndpoint() const
  254. {
  255. return ep;
  256. }
  257. virtual unsigned queryPort() const
  258. {
  259. return port;
  260. }
  261. };
  262. class ProtocolQueryWorker : public CInterface, implements IPooledThread
  263. {
  264. public:
  265. IMPLEMENT_IINTERFACE;
  266. ProtocolQueryWorker(ProtocolListener *_listener) : listener(_listener)
  267. {
  268. qstart = msTick();
  269. time(&startTime);
  270. }
  271. // interface IPooledThread
  272. virtual void init(void *)
  273. {
  274. qstart = msTick();
  275. time(&startTime);
  276. }
  277. virtual bool canReuse()
  278. {
  279. return true;
  280. }
  281. virtual bool stop()
  282. {
  283. ERRLOG("RoxieQueryWorker stopped with queries active");
  284. return true;
  285. }
  286. protected:
  287. ProtocolListener *listener;
  288. unsigned qstart;
  289. time_t startTime;
  290. };
  291. enum class AdaptiveRoot {NamedArray, RootArray, FirstRow};
  292. class AdaptiveRESTJsonWriter : public CommonJsonWriter
  293. {
  294. AdaptiveRoot model;
  295. unsigned depth = 0;
  296. public:
  297. AdaptiveRESTJsonWriter(AdaptiveRoot _model, unsigned _flags, unsigned _initialIndent, IXmlStreamFlusher *_flusher) :
  298. CommonJsonWriter(_flags, _initialIndent, _flusher), model(_model)
  299. {
  300. }
  301. virtual void outputBeginArray(const char *fieldname)
  302. {
  303. prepareBeginArray(fieldname);
  304. if (model == AdaptiveRoot::NamedArray || arrays.length()>1)
  305. appendJSONName(out, fieldname).append('[');
  306. else if (model == AdaptiveRoot::RootArray)
  307. out.append('[');
  308. }
  309. void outputEndArray(const char *fieldname)
  310. {
  311. arrays.pop();
  312. checkFormat(false, true, -1);
  313. if (arrays.length() || model != AdaptiveRoot::FirstRow)
  314. out.append(']');
  315. const char * sep = (fieldname) ? strchr(fieldname, '/') : NULL;
  316. while (sep)
  317. {
  318. out.append('}');
  319. sep = strchr(sep+1, '/');
  320. }
  321. }
  322. void outputBeginNested(const char *fieldname, bool nestChildren)
  323. {
  324. CommonJsonWriter::outputBeginNested(fieldname, nestChildren);
  325. if (model == AdaptiveRoot::FirstRow)
  326. depth++;
  327. }
  328. void outputEndNested(const char *fieldname)
  329. {
  330. CommonJsonWriter::outputEndNested(fieldname);
  331. if (model == AdaptiveRoot::FirstRow)
  332. {
  333. depth--;
  334. if (fieldname && streq(fieldname, "Row") && depth==0)
  335. {
  336. flush(true);
  337. flusher = nullptr;
  338. }
  339. }
  340. }
  341. };
  342. class AdaptiveRESTXmlWriter : public CommonXmlWriter
  343. {
  344. StringAttr tag;
  345. AdaptiveRoot model = AdaptiveRoot::NamedArray;
  346. unsigned depth = 0;
  347. public:
  348. AdaptiveRESTXmlWriter(AdaptiveRoot _model, const char *tagname, unsigned _flags, unsigned _initialIndent, IXmlStreamFlusher *_flusher) :
  349. CommonXmlWriter(_flags, _initialIndent, _flusher), tag(tagname), model(_model)
  350. {
  351. }
  352. void outputBeginNested(const char *fieldname, bool nestChildren)
  353. {
  354. if (model == AdaptiveRoot::FirstRow)
  355. {
  356. if (!depth && tag.length())
  357. fieldname = tag.str();
  358. depth++;
  359. }
  360. CommonXmlWriter::outputBeginNested(fieldname, nestChildren);
  361. }
  362. void outputEndNested(const char *fieldname)
  363. {
  364. if (model == AdaptiveRoot::FirstRow)
  365. {
  366. depth--;
  367. if (!depth)
  368. {
  369. CommonXmlWriter::outputEndNested(tag.length() ? tag.str() : fieldname);
  370. flush(true);
  371. flusher = nullptr;
  372. return;
  373. }
  374. }
  375. CommonXmlWriter::outputEndNested(fieldname);
  376. }
  377. };
  378. IXmlWriterExt * createAdaptiveRESTWriterExt(AdaptiveRoot model, const char *tagname, unsigned _flags, unsigned _initialIndent, IXmlStreamFlusher *_flusher, XMLWriterType xmlType)
  379. {
  380. if (xmlType==WTJSON)
  381. return new AdaptiveRESTJsonWriter(model, _flags, _initialIndent, _flusher);
  382. return new AdaptiveRESTXmlWriter(model, tagname, _flags, _initialIndent, _flusher);
  383. }
  384. //================================================================================================================
  385. class CHpccNativeResultsWriter : public CInterface, implements IHpccNativeProtocolResultsWriter
  386. {
  387. protected:
  388. SafeSocket *client;
  389. CriticalSection resultsCrit;
  390. IPointerArrayOf<FlushingStringBuffer> resultMap;
  391. StringAttr queryName;
  392. StringAttr tagName;
  393. StringAttr resultFilter;
  394. const IContextLogger &logctx;
  395. Owned<FlushingStringBuffer> probe;
  396. TextMarkupFormat mlFmt;
  397. PTreeReaderOptions xmlReadFlags;
  398. bool isBlocked;
  399. bool isRaw;
  400. bool isHTTP;
  401. bool trim;
  402. bool failed;
  403. bool adaptiveRoot = false;
  404. bool onlyUseFirstRow = false;
  405. public:
  406. IMPLEMENT_IINTERFACE;
  407. CHpccNativeResultsWriter(const char *queryname, SafeSocket *_client, bool _isBlocked, TextMarkupFormat _mlFmt, bool _isRaw, bool _isHTTP, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags) :
  408. client(_client), queryName(queryname), logctx(_logctx), mlFmt(_mlFmt), xmlReadFlags(_xmlReadFlags), isBlocked(_isBlocked), isRaw(_isRaw), isHTTP(_isHTTP)
  409. {
  410. }
  411. ~CHpccNativeResultsWriter()
  412. {
  413. }
  414. inline void setAdaptiveRoot(){adaptiveRoot = true; client->setAdaptiveRoot(true);}
  415. inline void setTagName(const char *tag){tagName.set(tag);}
  416. inline void setOnlyUseFirstRow(){onlyUseFirstRow = true;}
  417. inline void setResultFilter(const char *_resultFilter){resultFilter.set(_resultFilter);}
  418. virtual FlushingStringBuffer *queryResult(unsigned sequence)
  419. {
  420. CriticalBlock procedure(resultsCrit);
  421. while (!resultMap.isItem(sequence))
  422. resultMap.append(NULL);
  423. FlushingStringBuffer *result = resultMap.item(sequence);
  424. if (!result)
  425. {
  426. if (mlFmt==MarkupFmt_JSON)
  427. result = new FlushingJsonBuffer(client, isBlocked, isHTTP, logctx);
  428. else
  429. result = new FlushingStringBuffer(client, isBlocked, mlFmt, isRaw, isHTTP, logctx);
  430. result->isSoap = isHTTP;
  431. result->trim = trim;
  432. result->queryName.set(queryName);
  433. resultMap.replace(result, sequence);
  434. }
  435. return result;
  436. }
  437. virtual FlushingStringBuffer *createFlushingBuffer()
  438. {
  439. return new FlushingStringBuffer(client, isBlocked, mlFmt, isRaw, isHTTP, logctx);
  440. }
  441. bool checkAdaptiveResult(const char *name)
  442. {
  443. if (!adaptiveRoot)
  444. return false;
  445. if (!resultFilter || !*resultFilter)
  446. return true;
  447. return (streq(resultFilter, name));
  448. }
  449. virtual IXmlWriter *addDataset(const char *name, unsigned sequence, const char *elementName, bool &appendRawData, unsigned writeFlags, bool _extend, const IProperties *xmlns)
  450. {
  451. FlushingStringBuffer *response = queryResult(sequence);
  452. if (response)
  453. {
  454. appendRawData = response->isRaw;
  455. bool adaptive = checkAdaptiveResult(name);
  456. if (adaptive)
  457. {
  458. elementName = nullptr;
  459. if (response->mlFmt!=MarkupFmt_JSON && !onlyUseFirstRow && tagName.length())
  460. elementName = tagName.str();
  461. }
  462. response->startDataset(elementName, name, sequence, _extend, xmlns, adaptive);
  463. if (response->mlFmt==MarkupFmt_XML || response->mlFmt==MarkupFmt_JSON)
  464. {
  465. if (response->mlFmt==MarkupFmt_JSON)
  466. writeFlags |= XWFnoindent;
  467. AdaptiveRoot rootType = AdaptiveRoot::NamedArray;
  468. if (adaptive)
  469. {
  470. if (onlyUseFirstRow)
  471. rootType = AdaptiveRoot::FirstRow;
  472. else
  473. rootType = AdaptiveRoot::RootArray;
  474. }
  475. Owned<IXmlWriter> xmlwriter = createAdaptiveRESTWriterExt(rootType, tagName, writeFlags, 1, response, (response->mlFmt==MarkupFmt_JSON) ? WTJSON : WTStandard);
  476. xmlwriter->outputBeginArray("Row");
  477. return xmlwriter.getClear();
  478. }
  479. }
  480. return NULL;
  481. }
  482. virtual void finalizeXmlRow(unsigned sequence)
  483. {
  484. if (mlFmt==MarkupFmt_XML || mlFmt==MarkupFmt_JSON)
  485. {
  486. FlushingStringBuffer *r = queryResult(sequence);
  487. if (r)
  488. {
  489. r->incrementRowCount();
  490. r->flush(false);
  491. }
  492. }
  493. }
  494. inline void startScalar(FlushingStringBuffer *r, const char *name, unsigned sequence)
  495. {
  496. if (checkAdaptiveResult(name))
  497. {
  498. r->startScalar(name, sequence, true, tagName.length() ? tagName.str() : name);
  499. return;
  500. }
  501. r->startScalar(name, sequence);
  502. }
  503. virtual void appendRaw(unsigned sequence, unsigned len, const char *data)
  504. {
  505. FlushingStringBuffer *r = queryResult(sequence);
  506. if (r)
  507. r->append(len, data);
  508. }
  509. virtual void appendRawRow(unsigned sequence, unsigned len, const char *data)
  510. {
  511. FlushingStringBuffer *r = queryResult(sequence);
  512. if (r)
  513. {
  514. r->append(len, data);
  515. r->incrementRowCount();
  516. r->flush(false);
  517. }
  518. }
  519. virtual void appendSimpleRow(unsigned sequence, const char *str)
  520. {
  521. FlushingStringBuffer *r = queryResult(sequence);
  522. if (r)
  523. r->append(str);
  524. }
  525. virtual void setResultBool(const char *name, unsigned sequence, bool value)
  526. {
  527. FlushingStringBuffer *r = queryResult(sequence);
  528. if (r)
  529. {
  530. startScalar(r, name, sequence);
  531. if (isRaw)
  532. r->append(sizeof(value), (char *)&value);
  533. else
  534. r->append(value ? "true" : "false");
  535. }
  536. }
  537. virtual void setResultData(const char *name, unsigned sequence, int len, const void * data)
  538. {
  539. FlushingStringBuffer *r = queryResult(sequence);
  540. if (r)
  541. {
  542. startScalar(r, name, sequence);
  543. r->encodeData(data, len);
  544. }
  545. }
  546. virtual void setResultRaw(const char *name, unsigned sequence, int len, const void * data)
  547. {
  548. FlushingStringBuffer *r = queryResult(sequence);
  549. if (r)
  550. {
  551. startScalar(r, name, sequence);
  552. if (isRaw)
  553. r->append(len, (const char *) data);
  554. else
  555. UNIMPLEMENTED;
  556. }
  557. }
  558. virtual void setResultSet(const char *name, unsigned sequence, bool isAll, size32_t len, const void * data, ISetToXmlTransformer * transformer)
  559. {
  560. FlushingStringBuffer *r = queryResult(sequence);
  561. if (r)
  562. {
  563. startScalar(r, name, sequence);
  564. if (isRaw)
  565. r->append(len, (char *)data);
  566. else if (mlFmt==MarkupFmt_XML)
  567. {
  568. assertex(transformer);
  569. CommonXmlWriter writer(xmlReadFlags|XWFnoindent, 0);
  570. transformer->toXML(isAll, len, (byte *)data, writer);
  571. r->append(writer.str());
  572. }
  573. else if (mlFmt==MarkupFmt_JSON)
  574. {
  575. assertex(transformer);
  576. CommonJsonWriter writer(xmlReadFlags|XWFnoindent, 0);
  577. transformer->toXML(isAll, len, (byte *)data, writer);
  578. r->append(writer.str());
  579. }
  580. else
  581. {
  582. assertex(transformer);
  583. r->append('[');
  584. if (isAll)
  585. r->appendf("*]");
  586. else
  587. {
  588. SimpleOutputWriter x;
  589. transformer->toXML(isAll, len, (const byte *) data, x);
  590. r->appendf("%s]", x.str());
  591. }
  592. }
  593. }
  594. }
  595. virtual void setResultDecimal(const char *name, unsigned sequence, int len, int precision, bool isSigned, const void *val)
  596. {
  597. FlushingStringBuffer *r = queryResult(sequence);
  598. if (r)
  599. {
  600. startScalar(r, name, sequence);
  601. if (isRaw)
  602. r->append(len, (char *)val);
  603. else
  604. {
  605. StringBuffer s;
  606. if (isSigned)
  607. outputXmlDecimal(val, len, precision, NULL, s);
  608. else
  609. outputXmlUDecimal(val, len, precision, NULL, s);
  610. r->append(s);
  611. }
  612. }
  613. }
  614. virtual void setResultInt(const char *name, unsigned sequence, __int64 value, unsigned size)
  615. {
  616. FlushingStringBuffer *r = queryResult(sequence);
  617. if (r)
  618. {
  619. if (isRaw)
  620. {
  621. startScalar(r, name, sequence);
  622. r->append(sizeof(value), (char *)&value);
  623. }
  624. else
  625. r->setScalarInt(name, sequence, value, size);
  626. }
  627. }
  628. virtual void setResultUInt(const char *name, unsigned sequence, unsigned __int64 value, unsigned size)
  629. {
  630. FlushingStringBuffer *r = queryResult(sequence);
  631. if (r)
  632. {
  633. if (isRaw)
  634. {
  635. startScalar(r, name, sequence);
  636. r->append(sizeof(value), (char *)&value);
  637. }
  638. else
  639. r->setScalarUInt(name, sequence, value, size);
  640. }
  641. }
  642. virtual void setResultReal(const char *name, unsigned sequence, double value)
  643. {
  644. FlushingStringBuffer *r = queryResult(sequence);
  645. if (r)
  646. {
  647. startScalar(r, name, sequence);
  648. r->append(value);
  649. }
  650. }
  651. virtual void setResultString(const char *name, unsigned sequence, int len, const char * str)
  652. {
  653. FlushingStringBuffer *r = queryResult(sequence);
  654. if (r)
  655. {
  656. startScalar(r, name, sequence);
  657. if (r->isRaw)
  658. {
  659. r->append(len, str);
  660. }
  661. else
  662. {
  663. r->encodeString(str, len);
  664. }
  665. }
  666. }
  667. virtual void setResultUnicode(const char *name, unsigned sequence, int len, UChar const * str)
  668. {
  669. FlushingStringBuffer *r = queryResult(sequence);
  670. if (r)
  671. {
  672. startScalar(r, name, sequence);
  673. if (r->isRaw)
  674. {
  675. r->append(len*2, (const char *) str);
  676. }
  677. else
  678. {
  679. rtlDataAttr buff;
  680. unsigned bufflen = 0;
  681. rtlUnicodeToCodepageX(bufflen, buff.refstr(), len, str, "utf-8");
  682. r->encodeString(buff.getstr(), bufflen, true); // output as UTF-8
  683. }
  684. }
  685. }
  686. virtual void setResultVarString(const char * name, unsigned sequence, const char * value)
  687. {
  688. setResultString(name, sequence, strlen(value), value);
  689. }
  690. virtual void setResultVarUnicode(const char * name, unsigned sequence, UChar const * value)
  691. {
  692. setResultUnicode(name, sequence, rtlUnicodeStrlen(value), value);
  693. }
  694. virtual void flush()
  695. {
  696. ForEachItemIn(seq, resultMap)
  697. {
  698. FlushingStringBuffer *result = resultMap.item(seq);
  699. if (result)
  700. result->flush(true);
  701. }
  702. }
  703. virtual void finalize(unsigned seqNo, const char *delim, const char *filter)
  704. {
  705. bool needDelimiter = false;
  706. ForEachItemIn(seq, resultMap)
  707. {
  708. FlushingStringBuffer *result = resultMap.item(seq);
  709. if (result && (!filter || !*filter || streq(filter, result->queryResultName())))
  710. {
  711. result->flush(true);
  712. for(;;)
  713. {
  714. size32_t length;
  715. void *payload = result->getPayload(length);
  716. if (!length)
  717. break;
  718. if (needDelimiter)
  719. {
  720. StringAttr s(delim); //write() will take ownership of buffer
  721. size32_t len = s.length();
  722. client->write((void *)s.detach(), len, true);
  723. needDelimiter=false;
  724. }
  725. client->write(payload, length, true);
  726. }
  727. if (delim)
  728. needDelimiter=true;
  729. }
  730. }
  731. }
  732. virtual void appendProbeGraph(const char *xml)
  733. {
  734. if (!xml)
  735. {
  736. if (probe)
  737. probe.clear();
  738. return;
  739. }
  740. if (!probe)
  741. {
  742. probe.setown(new FlushingStringBuffer(client, isBlocked, MarkupFmt_XML, false, isHTTP, logctx));
  743. probe->startDataset("_Probe", NULL, (unsigned) -1); // initialize it
  744. }
  745. probe->append("\n");
  746. probe->append(xml);
  747. }
  748. };
  749. class CHpccXmlResultsWriter : public CHpccNativeResultsWriter
  750. {
  751. public:
  752. CHpccXmlResultsWriter(const char *queryname, SafeSocket *_client, bool _isHTTP, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags) :
  753. CHpccNativeResultsWriter(queryname, _client, false, MarkupFmt_XML, false, _isHTTP, _logctx, _xmlReadFlags)
  754. {
  755. }
  756. virtual void addContent(TextMarkupFormat fmt, const char *content, const char *name)
  757. {
  758. StringBuffer xml;
  759. if (!content || !*content)
  760. return;
  761. if (fmt==MarkupFmt_JSON)
  762. {
  763. Owned<IPropertyTree> convertPT = createPTreeFromXMLString(content);
  764. if (name && *name)
  765. appendXMLOpenTag(xml, name);
  766. toXML(convertPT, xml, 0, 0);
  767. if (name && *name)
  768. appendXMLCloseTag(xml, name);
  769. }
  770. }
  771. virtual void finalize(unsigned seqNo)
  772. {
  773. if (!isHTTP)
  774. {
  775. flush();
  776. return;
  777. }
  778. CriticalBlock b(resultsCrit);
  779. CriticalBlock b1(client->queryCrit());
  780. StringBuffer responseHead, responseTail;
  781. responseHead.append("<Results><Result>");
  782. unsigned len = responseHead.length();
  783. client->write(responseHead.detach(), len, true);
  784. ForEachItemIn(seq, resultMap)
  785. {
  786. FlushingStringBuffer *result = resultMap.item(seq);
  787. if (result)
  788. {
  789. result->flush(true);
  790. for(;;)
  791. {
  792. size32_t length;
  793. void *payload = result->getPayload(length);
  794. if (!length)
  795. break;
  796. client->write(payload, length, true);
  797. }
  798. }
  799. }
  800. responseTail.append("</Result></Results>");
  801. len = responseTail.length();
  802. client->write(responseTail.detach(), len, true);
  803. }
  804. };
  805. class CHpccJsonResultsWriter : public CHpccNativeResultsWriter
  806. {
  807. public:
  808. CHpccJsonResultsWriter(const char *queryname, SafeSocket *_client, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags) :
  809. CHpccNativeResultsWriter(queryname, _client, false, MarkupFmt_JSON, false, true, _logctx, _xmlReadFlags)
  810. {
  811. }
  812. virtual FlushingStringBuffer *createFlushingBuffer()
  813. {
  814. return new FlushingJsonBuffer(client, isBlocked, isHTTP, logctx);
  815. }
  816. virtual void finalize(unsigned seqNo)
  817. {
  818. CriticalBlock b(resultsCrit);
  819. CriticalBlock b1(client->queryCrit());
  820. StringBuffer responseHead, responseTail;
  821. appendJSONName(responseHead, "Results").append(" {");
  822. unsigned len = responseHead.length();
  823. client->write(responseHead.detach(), len, true);
  824. bool needDelimiter = false;
  825. ForEachItemIn(seq, resultMap)
  826. {
  827. FlushingStringBuffer *result = resultMap.item(seq);
  828. if (result)
  829. {
  830. result->flush(true);
  831. for(;;)
  832. {
  833. size32_t length;
  834. void *payload = result->getPayload(length);
  835. if (!length)
  836. break;
  837. if (needDelimiter)
  838. {
  839. StringAttr s(","); //write() will take ownership of buffer
  840. size32_t len = s.length();
  841. client->write((void *)s.detach(), len, true);
  842. needDelimiter=false;
  843. }
  844. client->write(payload, length, true);
  845. }
  846. needDelimiter=true;
  847. }
  848. }
  849. responseTail.append("}");
  850. len = responseTail.length();
  851. client->write(responseTail.detach(), len, true);
  852. }
  853. };
  854. class CHpccNativeProtocolResponse : public CInterface, implements IHpccNativeProtocolResponse
  855. {
  856. protected:
  857. SafeSocket *client;
  858. StringAttr queryName;
  859. StringArray resultFilter;
  860. StringBuffer rootTag;
  861. const IContextLogger &logctx;
  862. TextMarkupFormat mlFmt;
  863. PTreeReaderOptions xmlReadFlags;
  864. Owned<CHpccNativeResultsWriter> results; //hpcc results section
  865. IPointerArrayOf<FlushingStringBuffer> contentsMap; //other sections
  866. CriticalSection contentsCrit;
  867. unsigned protocolFlags;
  868. bool isHTTP;
  869. bool failed;
  870. public:
  871. IMPLEMENT_IINTERFACE;
  872. CHpccNativeProtocolResponse(const char *queryname, SafeSocket *_client, TextMarkupFormat _mlFmt, unsigned flags, bool _isHTTP, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags, const char *_resultFilterString, const char *_rootTag) :
  873. client(_client), queryName(queryname), logctx(_logctx), mlFmt(_mlFmt), xmlReadFlags(_xmlReadFlags), protocolFlags(flags), isHTTP(_isHTTP), rootTag(_rootTag)
  874. {
  875. resultFilter.appendList(_resultFilterString, ".");
  876. if (!rootTag.length() && resultFilter.length())
  877. rootTag.set(resultFilter.item(0)).replace(' ', '_');
  878. }
  879. ~CHpccNativeProtocolResponse()
  880. {
  881. }
  882. virtual unsigned getFlags()
  883. {
  884. return protocolFlags;
  885. }
  886. inline bool getIsRaw()
  887. {
  888. return (protocolFlags & HPCC_PROTOCOL_NATIVE_RAW);
  889. }
  890. inline bool getIsBlocked()
  891. {
  892. return (protocolFlags & HPCC_PROTOCOL_BLOCKED);
  893. }
  894. inline bool getTrim()
  895. {
  896. return (protocolFlags & HPCC_PROTOCOL_TRIM);
  897. }
  898. virtual FlushingStringBuffer *queryAppendContentBuffer()
  899. {
  900. CriticalBlock procedure(contentsCrit);
  901. FlushingStringBuffer *content;
  902. if (mlFmt==MarkupFmt_JSON)
  903. content = new FlushingJsonBuffer(client, getIsBlocked(), isHTTP, logctx);
  904. else
  905. content = new FlushingStringBuffer(client, getIsBlocked(), mlFmt, getIsRaw(), isHTTP, logctx);
  906. content->isSoap = isHTTP;
  907. content->trim = getTrim();
  908. content->queryName.set(queryName);
  909. if (!isHTTP)
  910. content->startBlock();
  911. contentsMap.append(content);
  912. return content;
  913. }
  914. virtual IHpccProtocolResultsWriter *queryHpccResultsSection()
  915. {
  916. if (!results)
  917. {
  918. results.setown(new CHpccNativeResultsWriter(queryName, client, getIsBlocked(), mlFmt, getIsRaw(), isHTTP, logctx, xmlReadFlags));
  919. if (rootTag.length())
  920. results->setTagName(rootTag);
  921. if (resultFilter.length())
  922. {
  923. results->setAdaptiveRoot();
  924. results->setResultFilter(resultFilter.item(0));
  925. }
  926. if (resultFilter.isItem(1) && strieq("row", resultFilter.item(1)))
  927. results->setOnlyUseFirstRow();
  928. }
  929. return results;
  930. }
  931. virtual void appendContent(TextMarkupFormat mlFmt, const char *content, const char *name=NULL)
  932. {
  933. throwUnexpected();
  934. }
  935. virtual IXmlWriter *writeAppendContent(const char *name = NULL)
  936. {
  937. throwUnexpected();
  938. }
  939. virtual void finalize(unsigned seqNo)
  940. {
  941. flush();
  942. if (!isHTTP)
  943. {
  944. unsigned replyLen = 0;
  945. client->write(&replyLen, sizeof(replyLen));
  946. }
  947. }
  948. virtual bool checkConnection()
  949. {
  950. if (client)
  951. return client->checkConnection();
  952. else
  953. return true;
  954. }
  955. virtual void sendHeartBeat()
  956. {
  957. if (client)
  958. client->sendHeartBeat(logctx);
  959. }
  960. virtual SafeSocket *querySafeSocket()
  961. {
  962. return client;
  963. }
  964. virtual void flush()
  965. {
  966. if (results)
  967. results->flush();
  968. ForEachItemIn(i, contentsMap)
  969. contentsMap.item(i)->flush(true);
  970. }
  971. virtual void appendProbeGraph(const char *xml)
  972. {
  973. if (results)
  974. results->appendProbeGraph(xml);
  975. }
  976. };
  977. class CHpccJsonResponse : public CHpccNativeProtocolResponse
  978. {
  979. public:
  980. CHpccJsonResponse(const char *queryname, SafeSocket *_client, unsigned flags, bool _isHttp, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags, const char *_resultFilter, const char *_rootTag) :
  981. CHpccNativeProtocolResponse(queryname, _client, MarkupFmt_JSON, flags, _isHttp, _logctx, _xmlReadFlags, _resultFilter, _rootTag)
  982. {
  983. }
  984. virtual IHpccProtocolResultsWriter *getHpccResultsSection()
  985. {
  986. if (!results)
  987. results.setown(new CHpccJsonResultsWriter(queryName, client, logctx, xmlReadFlags));
  988. return results;
  989. }
  990. virtual void appendContent(TextMarkupFormat mlFmt, const char *content, const char *name=NULL)
  991. {
  992. if (mlFmt!=MarkupFmt_XML && mlFmt!=MarkupFmt_JSON)
  993. return;
  994. StringBuffer json;
  995. if (mlFmt==MarkupFmt_XML)
  996. {
  997. Owned<IPropertyTree> convertPT = createPTreeFromXMLString(StringBuffer("<Control>").append(content).append("</Control>"));
  998. toJSON(convertPT, json, 0, 0);
  999. content = json.str();
  1000. }
  1001. FlushingStringBuffer *contentBuffer = queryAppendContentBuffer();
  1002. StringBuffer tag;
  1003. if (name && *name)
  1004. appendJSONName(tag, name);
  1005. contentBuffer->append(tag);
  1006. contentBuffer->append(content);
  1007. }
  1008. virtual IXmlWriter *writeAppendContent(const char *name = NULL)
  1009. {
  1010. FlushingStringBuffer *content = queryAppendContentBuffer();
  1011. if (name && *name)
  1012. {
  1013. StringBuffer tag;
  1014. appendJSONName(tag, name);
  1015. content->append(tag);
  1016. }
  1017. Owned<IXmlWriter> xmlwriter = createIXmlWriterExt(XWFnoindent, 1, content, WTJSON);
  1018. return xmlwriter.getClear();
  1019. }
  1020. virtual void outputContent()
  1021. {
  1022. CriticalBlock b1(client->queryCrit());
  1023. bool needDelimiter = false;
  1024. ForEachItemIn(seq, contentsMap)
  1025. {
  1026. FlushingStringBuffer *content = contentsMap.item(seq);
  1027. if (content)
  1028. {
  1029. content->flush(true);
  1030. for(;;)
  1031. {
  1032. size32_t length;
  1033. void *payload = content->getPayload(length);
  1034. if (!length)
  1035. break;
  1036. if (needDelimiter)
  1037. {
  1038. StringAttr s(","); //write() will take ownership of buffer
  1039. size32_t len = s.length();
  1040. client->write((void *)s.detach(), len, true);
  1041. needDelimiter=false;
  1042. }
  1043. client->write(payload, length, true);
  1044. }
  1045. needDelimiter=true;
  1046. }
  1047. }
  1048. }
  1049. virtual void finalize(unsigned seqNo)
  1050. {
  1051. if (!isHTTP)
  1052. {
  1053. CHpccNativeProtocolResponse::finalize(seqNo);
  1054. return;
  1055. }
  1056. CriticalBlock b(contentsCrit);
  1057. StringBuffer responseHead, responseTail;
  1058. if (!resultFilter.ordinality() && !(protocolFlags & HPCC_PROTOCOL_CONTROL))
  1059. {
  1060. StringBuffer name(queryName.get());
  1061. if (isHTTP)
  1062. name.append("Response");
  1063. appendJSONName(responseHead, name.str()).append(" {");
  1064. appendJSONValue(responseHead, "sequence", seqNo);
  1065. if (contentsMap.length() || results)
  1066. delimitJSON(responseHead);
  1067. unsigned len = responseHead.length();
  1068. client->write(responseHead.detach(), len, true);
  1069. }
  1070. if (!resultFilter.ordinality())
  1071. outputContent();
  1072. if (results)
  1073. results->finalize(seqNo, ",", resultFilter.ordinality() ? resultFilter.item(0) : NULL);
  1074. if (!resultFilter.ordinality() && !(protocolFlags & HPCC_PROTOCOL_CONTROL))
  1075. {
  1076. responseTail.append("}");
  1077. unsigned len = responseTail.length();
  1078. client->write(responseTail.detach(), len, true);
  1079. }
  1080. }
  1081. };
  1082. class CHpccXmlResponse : public CHpccNativeProtocolResponse
  1083. {
  1084. public:
  1085. CHpccXmlResponse(const char *queryname, SafeSocket *_client, unsigned flags, bool _isHTTP, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags, const char *_resultFilter, const char *_rootTag) :
  1086. CHpccNativeProtocolResponse(queryname, _client, MarkupFmt_XML, flags, _isHTTP, _logctx, _xmlReadFlags, _resultFilter, _rootTag)
  1087. {
  1088. }
  1089. virtual IHpccProtocolResultsWriter *getHpccResultsSection()
  1090. {
  1091. if (!results)
  1092. results.setown(new CHpccXmlResultsWriter(queryName, client, isHTTP, logctx, xmlReadFlags));
  1093. return results;
  1094. }
  1095. virtual void appendContent(TextMarkupFormat mlFmt, const char *content, const char *name=NULL)
  1096. {
  1097. if (mlFmt!=MarkupFmt_XML && mlFmt!=MarkupFmt_JSON)
  1098. return;
  1099. StringBuffer xml;
  1100. if (mlFmt==MarkupFmt_JSON)
  1101. {
  1102. Owned<IPropertyTree> convertPT = createPTreeFromJSONString(content);
  1103. toXML(convertPT, xml, 0, 0);
  1104. content = xml.str();
  1105. }
  1106. FlushingStringBuffer *contentBuffer = queryAppendContentBuffer();
  1107. if (name && *name)
  1108. {
  1109. StringBuffer tag;
  1110. appendXMLOpenTag(tag, name);
  1111. contentBuffer->append(tag.append('\n'));
  1112. appendXMLCloseTag(tag.clear(), name);
  1113. contentBuffer->setTail(tag.append('\n'));
  1114. }
  1115. contentBuffer->append(content);
  1116. }
  1117. virtual IXmlWriter *writeAppendContent(const char *name = NULL)
  1118. {
  1119. FlushingStringBuffer *content = queryAppendContentBuffer();
  1120. StringBuffer tag;
  1121. if (name && *name)
  1122. {
  1123. appendXMLOpenTag(tag, name);
  1124. content->append(tag);
  1125. appendXMLCloseTag(tag.clear(), name);
  1126. content->setTail(tag);
  1127. }
  1128. Owned<IXmlWriter> xmlwriter = createIXmlWriterExt(0, 1, content, WTStandard);
  1129. return xmlwriter.getClear();
  1130. }
  1131. virtual void outputContent()
  1132. {
  1133. CriticalBlock b1(client->queryCrit());
  1134. bool needDelimiter = false;
  1135. ForEachItemIn(seq, contentsMap)
  1136. {
  1137. FlushingStringBuffer *content = contentsMap.item(seq);
  1138. if (content)
  1139. {
  1140. content->flush(true);
  1141. if (!this->isHTTP)
  1142. continue;
  1143. for(;;)
  1144. {
  1145. size32_t length;
  1146. void *payload = content->getPayload(length);
  1147. if (!length)
  1148. break;
  1149. client->write(payload, length, true);
  1150. }
  1151. }
  1152. }
  1153. }
  1154. virtual void finalize(unsigned seqNo)
  1155. {
  1156. if (!isHTTP)
  1157. {
  1158. CHpccNativeProtocolResponse::finalize(seqNo);
  1159. return;
  1160. }
  1161. CriticalBlock b(contentsCrit);
  1162. StringBuffer responseHead, responseTail;
  1163. if (!resultFilter.ordinality() && !(protocolFlags & HPCC_PROTOCOL_CONTROL))
  1164. {
  1165. responseHead.append("<").append(queryName);
  1166. responseHead.append("Response").append(" xmlns=\"urn:hpccsystems:ecl:").appendLower(queryName.length(), queryName.str()).append('\"');
  1167. responseHead.append(" sequence=\"").append(seqNo).append("\">");
  1168. unsigned len = responseHead.length();
  1169. client->write(responseHead.detach(), len, true);
  1170. }
  1171. if (!resultFilter.ordinality())
  1172. outputContent();
  1173. if (results)
  1174. results->finalize(seqNo, NULL, resultFilter.ordinality() ? resultFilter.item(0) : NULL);
  1175. if (!resultFilter.ordinality() && !(protocolFlags & HPCC_PROTOCOL_CONTROL))
  1176. {
  1177. responseTail.append("</").append(queryName);
  1178. if (isHTTP)
  1179. responseTail.append("Response");
  1180. responseTail.append('>');
  1181. unsigned len = responseTail.length();
  1182. client->write(responseTail.detach(), len, true);
  1183. }
  1184. }
  1185. };
  1186. IHpccProtocolResponse *createProtocolResponse(const char *queryname, SafeSocket *client, HttpHelper &httpHelper, const IContextLogger &logctx, unsigned protocolFlags, PTreeReaderOptions xmlReadFlags)
  1187. {
  1188. StringAttr filter, tag;
  1189. httpHelper.getResultFilterAndTag(filter, tag);
  1190. if (protocolFlags & HPCC_PROTOCOL_NATIVE_RAW || protocolFlags & HPCC_PROTOCOL_NATIVE_ASCII)
  1191. return new CHpccNativeProtocolResponse(queryname, client, MarkupFmt_Unknown, protocolFlags, false, logctx, xmlReadFlags, filter, tag);
  1192. else if (httpHelper.queryResponseMlFormat()==MarkupFmt_JSON)
  1193. return new CHpccJsonResponse(queryname, client, protocolFlags, httpHelper.isHttp(), logctx, xmlReadFlags, filter, tag);
  1194. return new CHpccXmlResponse(queryname, client, protocolFlags, httpHelper.isHttp(), logctx, xmlReadFlags, filter, tag);
  1195. }
  1196. class CHttpRequestAsyncFor : public CInterface, public CAsyncFor
  1197. {
  1198. private:
  1199. const char *queryName, *queryText, *querySetName;
  1200. const IContextLogger &logctx;
  1201. IArrayOf<IPropertyTree> &requestArray;
  1202. Linked<IHpccProtocolMsgSink> sink;
  1203. Linked<IHpccProtocolMsgContext> msgctx;
  1204. SafeSocket &client;
  1205. HttpHelper &httpHelper;
  1206. PTreeReaderOptions xmlReadFlags;
  1207. unsigned &memused;
  1208. unsigned &slaveReplyLen;
  1209. CriticalSection crit;
  1210. unsigned flags;
  1211. public:
  1212. CHttpRequestAsyncFor(const char *_queryName, IHpccProtocolMsgSink *_sink, IHpccProtocolMsgContext *_msgctx, IArrayOf<IPropertyTree> &_requestArray,
  1213. SafeSocket &_client, HttpHelper &_httpHelper, unsigned _flags, unsigned &_memused, unsigned &_slaveReplyLen, const char *_queryText, const IContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags, const char *_querySetName)
  1214. : sink(_sink), msgctx(_msgctx), requestArray(_requestArray), client(_client), httpHelper(_httpHelper), memused(_memused),
  1215. slaveReplyLen(_slaveReplyLen), logctx(_logctx), xmlReadFlags(_xmlReadFlags), querySetName(_querySetName), flags(_flags)
  1216. {
  1217. queryName = _queryName;
  1218. queryText = _queryText;
  1219. }
  1220. IMPLEMENT_IINTERFACE;
  1221. void onException(IException *E)
  1222. {
  1223. //if (!logctx.isBlind())
  1224. // logctx.CTXLOG("FAILED: %s", queryText);
  1225. StringBuffer error("EXCEPTION: ");
  1226. E->errorMessage(error);
  1227. DBGLOG("%s", error.str());
  1228. client.checkSendHttpException(httpHelper, E, queryName);
  1229. E->Release();
  1230. }
  1231. void Do(unsigned idx)
  1232. {
  1233. try
  1234. {
  1235. IPropertyTree &request = requestArray.item(idx);
  1236. Owned<IHpccProtocolResponse> protocol = createProtocolResponse(request.queryName(), &client, httpHelper, logctx, flags, xmlReadFlags);
  1237. sink->onQueryMsg(msgctx, &request, protocol, flags, xmlReadFlags, querySetName, idx, memused, slaveReplyLen);
  1238. }
  1239. catch (IException * E)
  1240. {
  1241. onException(E);
  1242. }
  1243. catch (...)
  1244. {
  1245. onException(MakeStringException(ROXIE_INTERNAL_ERROR, "Unknown exception"));
  1246. }
  1247. }
  1248. };
  1249. //ADF - Haven't changed it yet, but this should eliminate the need to parse the query twice below
  1250. //I can load the query and lookup the parse flags before doing a full parse
  1251. //if it turns out I need more info I may delete this.
  1252. class QueryNameExtractor : public CInterface, implements IPTreeNotifyEvent
  1253. {
  1254. public:
  1255. TextMarkupFormat mlFmt;
  1256. StringAttr prefix;
  1257. StringAttr name;
  1258. unsigned headerDepth;
  1259. bool isSoap;
  1260. bool isRequestArray;
  1261. bool stripWhitespace;
  1262. bool more;
  1263. public:
  1264. IMPLEMENT_IINTERFACE;
  1265. QueryNameExtractor(TextMarkupFormat _mlFmt, bool _stripWhitespace) : mlFmt(_mlFmt), headerDepth(0), isSoap(false), isRequestArray(false), stripWhitespace(_stripWhitespace), more(true)
  1266. {
  1267. }
  1268. void extractName(const char *msg, const IContextLogger &logctx, const char *peer, unsigned port)
  1269. {
  1270. Owned<IPullPTreeReader> parser;
  1271. if (mlFmt==MarkupFmt_JSON)
  1272. parser.setown(createPullJSONStringReader(msg, *this));
  1273. else if (mlFmt==MarkupFmt_XML)
  1274. parser.setown(createPullXMLStringReader(msg, *this));
  1275. if (!parser)
  1276. return;
  1277. while (more && parser->next());
  1278. if (name.isEmpty())
  1279. {
  1280. const char *fmt = mlFmt==MarkupFmt_XML ? "XML" : "JSON";
  1281. IException *E = MakeStringException(-1, "ERROR: Invalid %s received from %s:%d - %s queryName not found", fmt, peer, port, msg);
  1282. logctx.logOperatorException(E, __FILE__, __LINE__, "Invalid query %s", fmt);
  1283. throw E;
  1284. }
  1285. String nameStr(name.get());
  1286. if (nameStr.endsWith("RequestArray"))
  1287. {
  1288. isRequestArray = true;
  1289. name.set(nameStr.str(), nameStr.length() - strlen("RequestArray"));
  1290. }
  1291. else if (nameStr.endsWith("Request"))
  1292. {
  1293. name.set(nameStr.str(), nameStr.length() - strlen("Request"));
  1294. }
  1295. }
  1296. virtual void beginNode(const char *tag, offset_t startOffset)
  1297. {
  1298. if (streq(tag, "__object__"))
  1299. return;
  1300. const char *local = strchr(tag, ':');
  1301. if (local)
  1302. local++;
  1303. else
  1304. local = tag;
  1305. if (mlFmt==MarkupFmt_XML)
  1306. {
  1307. if (!isSoap && streq(local, "Envelope"))
  1308. {
  1309. isSoap=true;
  1310. return;
  1311. }
  1312. if (isSoap && streq(local, "Header"))
  1313. {
  1314. headerDepth++;
  1315. return;
  1316. }
  1317. if (isSoap && !headerDepth && streq(local, "Body"))
  1318. return;
  1319. }
  1320. if (!headerDepth)
  1321. {
  1322. name.set(local);
  1323. if (tag!=local)
  1324. prefix.set(tag, local-tag-1);
  1325. }
  1326. }
  1327. virtual void newAttribute(const char *attr, const char *value)
  1328. {
  1329. if (!name.isEmpty() && streq(attr, "@_stripWhitespaceFromStoredDataset"))
  1330. {
  1331. stripWhitespace = strToBool(value);
  1332. more = false;
  1333. }
  1334. }
  1335. virtual void beginNodeContent(const char *tag)
  1336. {
  1337. if (!name.isEmpty())
  1338. more = false;
  1339. }
  1340. virtual void endNode(const char *tag, unsigned length, const void *value, bool binary, offset_t endOffset)
  1341. {
  1342. if (!name.isEmpty())
  1343. more = false;
  1344. else if (headerDepth && streq(tag, "Header"))
  1345. headerDepth--;
  1346. }
  1347. };
  1348. static Owned<IActiveQueryLimiterFactory> queryLimiterFactory;
  1349. class RoxieSocketWorker : public ProtocolQueryWorker
  1350. {
  1351. SocketEndpoint ep;
  1352. Owned<SafeSocket> client;
  1353. Owned<IHpccNativeProtocolMsgSink> sink;
  1354. public:
  1355. IMPLEMENT_IINTERFACE;
  1356. RoxieSocketWorker(ProtocolSocketListener *_pool, SocketEndpoint &_ep)
  1357. : ProtocolQueryWorker(_pool), ep(_ep)
  1358. {
  1359. sink.set(dynamic_cast<IHpccNativeProtocolMsgSink*>(_pool->queryMsgSink()));
  1360. }
  1361. // interface IPooledThread
  1362. virtual void init(void *_r)
  1363. {
  1364. client.setown(new CSafeSocket((ISocket *) _r));
  1365. ProtocolQueryWorker::init(_r);
  1366. }
  1367. virtual void main()
  1368. {
  1369. doMain("");
  1370. }
  1371. virtual void runOnce(const char *query)
  1372. {
  1373. doMain(query);
  1374. }
  1375. private:
  1376. static void sendHttpServerTooBusy(SafeSocket &client, const IContextLogger &logctx)
  1377. {
  1378. StringBuffer message;
  1379. message.append("HTTP/1.0 503 Server Too Busy\r\n\r\n");
  1380. message.append("Server too busy, please try again later");
  1381. StringBuffer err("Too many active queries"); // write out Too many active queries - make searching for this error consistent
  1382. if (!global->trapTooManyActiveQueries)
  1383. {
  1384. err.appendf(" %s", message.str());
  1385. logctx.CTXLOG("%s", err.str());
  1386. }
  1387. else
  1388. {
  1389. IException *E = MakeStringException(ROXIE_TOO_MANY_QUERIES, "%s", err.str());
  1390. logctx.logOperatorException(E, __FILE__, __LINE__, "%s", message.str());
  1391. E->Release();
  1392. }
  1393. try
  1394. {
  1395. client.write(message.str(), message.length());
  1396. }
  1397. catch (IException *E)
  1398. {
  1399. logctx.logOperatorException(E, __FILE__, __LINE__, "Exception caught in sendHttpServerTooBusy");
  1400. E->Release();
  1401. }
  1402. catch (...)
  1403. {
  1404. logctx.logOperatorException(NULL, __FILE__, __LINE__, "sendHttpServerTooBusy write failed (Unknown exception)");
  1405. }
  1406. }
  1407. void skipProtocolRoot(Owned<IPropertyTree> &queryPT, HttpHelper &httpHelper, StringAttr &queryName, bool &isRequest, bool &isRequestArray)
  1408. {
  1409. if (queryPT)
  1410. {
  1411. queryName.set(queryPT->queryName());
  1412. isRequest = false;
  1413. isRequestArray = false;
  1414. if (httpHelper.isHttp())
  1415. {
  1416. if (httpHelper.queryRequestMlFormat()==MarkupFmt_JSON)
  1417. {
  1418. if (strieq(queryName, "__array__"))
  1419. throw MakeStringException(ROXIE_DATA_ERROR, "JSON request array not implemented");
  1420. isRequest=true;
  1421. if (strieq(queryName, "__object__"))
  1422. {
  1423. queryPT.setown(queryPT->getPropTree("*[1]"));
  1424. queryName.set(queryPT->queryName());
  1425. if (!queryPT)
  1426. throw MakeStringException(ROXIE_DATA_ERROR, "Malformed JSON request (missing Body)");
  1427. }
  1428. }
  1429. else
  1430. {
  1431. if (strieq(queryName, "envelope"))
  1432. queryPT.setown(queryPT->getPropTree("Body/*"));
  1433. else if (!strnicmp(httpHelper.queryContentType(), "application/soap", strlen("application/soap")))
  1434. throw MakeStringException(ROXIE_DATA_ERROR, "Malformed SOAP request");
  1435. if (!queryPT)
  1436. throw MakeStringException(ROXIE_DATA_ERROR, "Malformed SOAP request (missing Body)");
  1437. String reqName(queryPT->queryName());
  1438. queryPT->removeProp("@xmlns:m");
  1439. // following code is moved from main() - should be no performance hit
  1440. String requestString("Request");
  1441. String requestArrayString("RequestArray");
  1442. if (reqName.endsWith(requestArrayString))
  1443. {
  1444. isRequestArray = true;
  1445. queryName.set(reqName.str(), reqName.length() - requestArrayString.length());
  1446. }
  1447. else if (reqName.endsWith(requestString))
  1448. {
  1449. isRequest = true;
  1450. queryName.set(reqName.str(), reqName.length() - requestString.length());
  1451. }
  1452. else
  1453. queryName.set(reqName.str());
  1454. queryPT->renameProp("/", queryName.get()); // reset the name of the tree
  1455. }
  1456. }
  1457. }
  1458. }
  1459. void sanitizeQuery(Owned<IPropertyTree> &queryPT, StringAttr &queryName, StringBuffer &saniText, HttpHelper &httpHelper, const char *&uid, bool &isRequest, bool &isRequestArray, bool &isBlind, bool &isDebug)
  1460. {
  1461. if (queryPT)
  1462. {
  1463. skipProtocolRoot(queryPT, httpHelper, queryName, isRequest, isRequestArray);
  1464. // convert to XML with attribute values in single quotes - makes replaying queries easier
  1465. uid = queryPT->queryProp("@uid");
  1466. if (!uid)
  1467. uid = queryPT->queryProp("_TransactionId");
  1468. isBlind = queryPT->getPropBool("@blind", false) || queryPT->getPropBool("_blind", false);
  1469. isDebug = queryPT->getPropBool("@debug") || queryPT->getPropBool("_Probe", false);
  1470. toXML(queryPT, saniText, 0, isBlind ? (XML_SingleQuoteAttributeValues | XML_Sanitize) : XML_SingleQuoteAttributeValues);
  1471. }
  1472. else
  1473. throw MakeStringException(ROXIE_DATA_ERROR, "Malformed request");
  1474. }
  1475. void createQueryPTree(Owned<IPropertyTree> &queryPT, HttpHelper &httpHelper, const char *text, byte flags, PTreeReaderOptions options)
  1476. {
  1477. StringBuffer logxml;
  1478. if (httpHelper.queryRequestMlFormat()==MarkupFmt_URL)
  1479. {
  1480. queryPT.setown(httpHelper.createPTreeFromParameters(flags));
  1481. toXML(queryPT, logxml);
  1482. DBGLOG("%s", logxml.str());
  1483. return;
  1484. }
  1485. if (httpHelper.queryRequestMlFormat()==MarkupFmt_JSON)
  1486. queryPT.setown(createPTreeFromJSONString(text, flags, options));
  1487. else
  1488. queryPT.setown(createPTreeFromXMLString(text, flags, options));
  1489. queryPT.setown(httpHelper.checkAddWrapperForAdaptiveInput(queryPT.getClear(), flags));
  1490. }
  1491. void doMain(const char *runQuery)
  1492. {
  1493. StringBuffer rawText(runQuery);
  1494. unsigned memused = 0;
  1495. IpAddress peer;
  1496. bool continuationNeeded = false;
  1497. bool isStatus = false;
  1498. Owned<IHpccProtocolMsgContext> msgctx = sink->createMsgContext(startTime);
  1499. IContextLogger &logctx = *msgctx->queryLogContext();
  1500. readAnother:
  1501. unsigned slavesReplyLen = 0;
  1502. StringArray allTargets;
  1503. sink->getTargetNames(allTargets);
  1504. HttpHelper httpHelper(&allTargets);
  1505. try
  1506. {
  1507. if (client)
  1508. {
  1509. client->querySocket()->getPeerAddress(peer);
  1510. if (!client->readBlock(rawText.clear(), WAIT_FOREVER, &httpHelper, continuationNeeded, isStatus, global->maxBlockSize))
  1511. {
  1512. if (traceLevel > 8)
  1513. {
  1514. StringBuffer b;
  1515. DBGLOG("No data reading query from socket");
  1516. }
  1517. client.clear();
  1518. return;
  1519. }
  1520. }
  1521. if (continuationNeeded)
  1522. {
  1523. qstart = msTick();
  1524. time(&startTime);
  1525. }
  1526. }
  1527. catch (IException * E)
  1528. {
  1529. if (traceLevel > 0)
  1530. {
  1531. StringBuffer b;
  1532. DBGLOG("Error reading query from socket: %s", E->errorMessage(b).str());
  1533. }
  1534. E->Release();
  1535. client.clear();
  1536. return;
  1537. }
  1538. bool isHTTP = httpHelper.isHttp();
  1539. TextMarkupFormat mlResponseFmt = MarkupFmt_Unknown;
  1540. TextMarkupFormat mlRequestFmt = MarkupFmt_Unknown;
  1541. if (!isStatus)
  1542. {
  1543. if (!isHTTP)
  1544. mlResponseFmt = mlRequestFmt = MarkupFmt_XML;
  1545. else
  1546. {
  1547. mlResponseFmt = httpHelper.queryResponseMlFormat();
  1548. mlRequestFmt = httpHelper.queryRequestMlFormat();
  1549. }
  1550. }
  1551. bool failed = false;
  1552. bool isRequest = false;
  1553. bool isRequestArray = false;
  1554. bool isBlind = false;
  1555. bool isDebug = false;
  1556. unsigned protocolFlags = isHTTP ? 0 : HPCC_PROTOCOL_NATIVE;
  1557. Owned<IPropertyTree> queryPT;
  1558. StringBuffer sanitizedText;
  1559. StringBuffer peerStr;
  1560. peer.getIpText(peerStr);
  1561. const char *uid = "-";
  1562. StringAttr queryName;
  1563. StringAttr queryPrefix;
  1564. bool stripWhitespace = msgctx->getStripWhitespace();
  1565. try
  1566. {
  1567. if (httpHelper.isHttpGet() || httpHelper.isFormPost())
  1568. {
  1569. queryName.set(httpHelper.queryQueryName());
  1570. if (httpHelper.isControlUrl())
  1571. queryPrefix.set("control");
  1572. }
  1573. else if (mlRequestFmt==MarkupFmt_XML || mlRequestFmt==MarkupFmt_JSON)
  1574. {
  1575. QueryNameExtractor extractor(mlRequestFmt, stripWhitespace);
  1576. extractor.extractName(rawText.str(), logctx, peerStr, ep.port);
  1577. queryName.set(extractor.name);
  1578. queryPrefix.set(extractor.prefix);
  1579. stripWhitespace = extractor.stripWhitespace;
  1580. if (httpHelper.isHttp())
  1581. httpHelper.setUseEnvelope(extractor.isSoap);
  1582. }
  1583. if (streq(queryPrefix.str(), "control"))
  1584. {
  1585. if (httpHelper.isHttp())
  1586. client->setHttpMode(queryName, false, httpHelper);
  1587. bool aclupdate = strieq(queryName, "aclupdate"); //ugly
  1588. byte iptFlags = aclupdate ? ipt_caseInsensitive : 0;
  1589. createQueryPTree(queryPT, httpHelper, rawText, iptFlags, (PTreeReaderOptions)(ptr_ignoreWhiteSpace|ptr_ignoreNameSpaces));
  1590. IPropertyTree *root = queryPT;
  1591. skipProtocolRoot(queryPT, httpHelper, queryName, isRequest, isRequestArray);
  1592. if (!strchr(queryName, ':'))
  1593. {
  1594. VStringBuffer fullname("control:%s", queryName.str()); //just easier to keep for debugging and internal checking
  1595. queryPT->renameProp("/", fullname);
  1596. }
  1597. Owned<IHpccProtocolResponse> protocol = createProtocolResponse(queryPT->queryName(), client, httpHelper, logctx, protocolFlags | HPCC_PROTOCOL_CONTROL, global->defaultXmlReadFlags);
  1598. sink->onControlMsg(msgctx, queryPT, protocol);
  1599. protocol->finalize(0);
  1600. if (streq(queryName, "lock") || streq(queryName, "childlock"))
  1601. goto readAnother;
  1602. }
  1603. else if (isStatus)
  1604. {
  1605. client->write("OK", 2);
  1606. }
  1607. else
  1608. {
  1609. unsigned readFlags = (unsigned) global->defaultXmlReadFlags | ptr_ignoreNameSpaces;
  1610. readFlags &= ~ptr_ignoreWhiteSpace;
  1611. readFlags |= (stripWhitespace ? ptr_ignoreWhiteSpace : ptr_none);
  1612. try
  1613. {
  1614. createQueryPTree(queryPT, httpHelper, rawText.str(), ipt_caseInsensitive, (PTreeReaderOptions)readFlags);
  1615. }
  1616. catch (IException *E)
  1617. {
  1618. logctx.logOperatorException(E, __FILE__, __LINE__, "Invalid XML received from %s:%d - %s", peerStr.str(), listener->queryPort(), rawText.str());
  1619. logctx.CTXLOG("ERROR: Invalid XML received from %s:%d - %s", peerStr.str(), listener->queryPort(), rawText.str());
  1620. throw;
  1621. }
  1622. uid = NULL;
  1623. sanitizeQuery(queryPT, queryName, sanitizedText, httpHelper, uid, isRequest, isRequestArray, isBlind, isDebug);
  1624. if (!uid)
  1625. uid = "-";
  1626. sink->checkAccess(peer, queryName, sanitizedText, isBlind);
  1627. if (isDebug)
  1628. msgctx->verifyAllowDebug();
  1629. isBlind = msgctx->checkSetBlind(isBlind);
  1630. if (msgctx->logFullQueries())
  1631. {
  1632. StringBuffer soapStr;
  1633. (isRequest) ? soapStr.append("SoapRequest") : (isRequestArray) ? soapStr.append("SoapRequest") : soapStr.clear();
  1634. logctx.CTXLOG("%s %s:%d %s %s %s", isBlind ? "BLIND:" : "QUERY:", peerStr.str(), listener->queryPort(), uid, soapStr.str(), sanitizedText.str());
  1635. }
  1636. if (strieq(queryPrefix.str(), "debug"))
  1637. {
  1638. FlushingStringBuffer response(client, false, MarkupFmt_XML, false, isHTTP, logctx);
  1639. response.startDataset("Debug", NULL, (unsigned) -1);
  1640. CommonXmlWriter out(0, 1);
  1641. sink->onDebugMsg(msgctx, uid, queryPT, out);
  1642. response.append(out.str());
  1643. }
  1644. Owned<IActiveQueryLimiter> l;
  1645. if (queryLimiterFactory)
  1646. l.setown(queryLimiterFactory->create(listener));
  1647. if (l && !l->isAccepted())
  1648. {
  1649. if (isHTTP)
  1650. {
  1651. sendHttpServerTooBusy(*client, logctx);
  1652. logctx.CTXLOG("FAILED: %s", sanitizedText.str());
  1653. logctx.CTXLOG("EXCEPTION: Too many active queries");
  1654. }
  1655. else
  1656. {
  1657. IException *e = MakeStringException(ROXIE_TOO_MANY_QUERIES, "Too many active queries");
  1658. if (msgctx->trapTooManyActiveQueries())
  1659. logctx.logOperatorException(e, __FILE__, __LINE__, NULL);
  1660. throw e;
  1661. }
  1662. }
  1663. else
  1664. {
  1665. StringBuffer querySetName;
  1666. if (isHTTP)
  1667. {
  1668. client->setHttpMode(queryName, isRequestArray, httpHelper);
  1669. querySetName.set(httpHelper.queryTarget());
  1670. if (querySetName.length())
  1671. {
  1672. const char *target = global->targetAliases->queryProp(querySetName.str());
  1673. if (target)
  1674. querySetName.set(target);
  1675. }
  1676. }
  1677. if (msgctx->initQuery(querySetName, queryName))
  1678. {
  1679. int bindCores = queryPT->getPropInt("@bindCores", msgctx->getBindCores());
  1680. if (bindCores > 0)
  1681. listener->setThreadAffinity(bindCores);
  1682. IArrayOf<IPropertyTree> requestArray;
  1683. if (isHTTP)
  1684. {
  1685. if (isRequestArray)
  1686. {
  1687. StringBuffer reqIterString;
  1688. reqIterString.append(queryName).append("Request");
  1689. Owned<IPropertyTreeIterator> reqIter = queryPT->getElements(reqIterString.str());
  1690. ForEach(*reqIter)
  1691. {
  1692. IPropertyTree *fixedreq = createPTree(queryName, ipt_caseInsensitive);
  1693. Owned<IPropertyTreeIterator> iter = reqIter->query().getElements("*");
  1694. ForEach(*iter)
  1695. {
  1696. fixedreq->addPropTree(iter->query().queryName(), LINK(&iter->query()));
  1697. }
  1698. requestArray.append(*fixedreq);
  1699. }
  1700. }
  1701. else
  1702. {
  1703. IPropertyTree *fixedreq = createPTree(queryName, ipt_caseInsensitive);
  1704. Owned<IPropertyTreeIterator> iter = queryPT->getElements("*");
  1705. ForEach(*iter)
  1706. {
  1707. fixedreq->addPropTree(iter->query().queryName(), LINK(&iter->query()));
  1708. }
  1709. requestArray.append(*fixedreq);
  1710. }
  1711. if (httpHelper.getTrim())
  1712. protocolFlags |= HPCC_PROTOCOL_TRIM;
  1713. }
  1714. else
  1715. {
  1716. const char *format = queryPT->queryProp("@format");
  1717. if (format)
  1718. {
  1719. if (stricmp(format, "raw") == 0)
  1720. {
  1721. protocolFlags |= HPCC_PROTOCOL_NATIVE_RAW;
  1722. mlResponseFmt = MarkupFmt_Unknown;
  1723. }
  1724. else if (stricmp(format, "bxml") == 0)
  1725. {
  1726. protocolFlags |= HPCC_PROTOCOL_BLOCKED;
  1727. }
  1728. else if (stricmp(format, "ascii") == 0)
  1729. {
  1730. protocolFlags |= HPCC_PROTOCOL_NATIVE_ASCII;
  1731. mlResponseFmt = MarkupFmt_Unknown;
  1732. }
  1733. else if (stricmp(format, "xml") != 0) // xml is the default
  1734. throw MakeStringException(ROXIE_INVALID_INPUT, "Unsupported format specified: %s", format);
  1735. }
  1736. if (queryPT->getPropBool("@trim", false))
  1737. protocolFlags |= HPCC_PROTOCOL_TRIM;
  1738. msgctx->setIntercept(queryPT->getPropBool("@log", false));
  1739. msgctx->setTraceLevel(queryPT->getPropInt("@traceLevel", logctx.queryTraceLevel()));
  1740. }
  1741. msgctx->noteQueryActive();
  1742. if (isHTTP)
  1743. {
  1744. CHttpRequestAsyncFor af(queryName, sink, msgctx, requestArray, *client, httpHelper, protocolFlags, memused, slavesReplyLen, sanitizedText, logctx, (PTreeReaderOptions)readFlags, querySetName);
  1745. af.For(requestArray.length(), global->numRequestArrayThreads);
  1746. }
  1747. else
  1748. {
  1749. Owned<IHpccProtocolResponse> protocol = createProtocolResponse(queryPT->queryName(), client, httpHelper, logctx, protocolFlags, (PTreeReaderOptions)readFlags);
  1750. sink->onQueryMsg(msgctx, queryPT, protocol, protocolFlags, (PTreeReaderOptions)readFlags, querySetName, 0, memused, slavesReplyLen);
  1751. }
  1752. }
  1753. }
  1754. }
  1755. }
  1756. catch (IException * E)
  1757. {
  1758. failed = true;
  1759. logctx.CTXLOG("FAILED: %s", sanitizedText.str());
  1760. StringBuffer error;
  1761. E->errorMessage(error);
  1762. logctx.CTXLOG("EXCEPTION: %s", error.str());
  1763. unsigned code = E->errorCode();
  1764. if (QUERYINTERFACE(E, ISEH_Exception))
  1765. code = ROXIE_INTERNAL_ERROR;
  1766. else if (QUERYINTERFACE(E, IOutOfMemException))
  1767. code = ROXIE_MEMORY_ERROR;
  1768. if (client)
  1769. {
  1770. if (isHTTP)
  1771. client->checkSendHttpException(httpHelper, E, queryName);
  1772. else
  1773. client->sendException("Roxie", code, error.str(), (protocolFlags & HPCC_PROTOCOL_NATIVE_RAW), logctx);
  1774. }
  1775. else
  1776. {
  1777. fprintf(stderr, "EXCEPTION: %s\n", error.str());
  1778. }
  1779. E->Release();
  1780. }
  1781. #ifndef _DEBUG
  1782. catch(...)
  1783. {
  1784. failed = true;
  1785. logctx.CTXLOG("FAILED: %s", sanitizedText.str());
  1786. logctx.CTXLOG("EXCEPTION: Unknown exception");
  1787. {
  1788. if (isHTTP)
  1789. {
  1790. Owned<IException> E = MakeStringException(ROXIE_INTERNAL_ERROR, "Unknown exception");
  1791. client->checkSendHttpException(httpHelper, E, queryName);
  1792. }
  1793. else
  1794. client->sendException("Roxie", ROXIE_INTERNAL_ERROR, "Unknown exception", (protocolFlags & HPCC_PROTOCOL_BLOCKED), logctx);
  1795. }
  1796. }
  1797. #endif
  1798. if (isHTTP)
  1799. {
  1800. try
  1801. {
  1802. client->flush();
  1803. }
  1804. catch (IException * E)
  1805. {
  1806. StringBuffer error("RoxieSocketWorker failed to write to socket ");
  1807. E->errorMessage(error);
  1808. logctx.CTXLOG("%s", error.str());
  1809. E->Release();
  1810. }
  1811. catch(...)
  1812. {
  1813. logctx.CTXLOG("RoxieSocketWorker failed to write to socket (Unknown exception)");
  1814. }
  1815. }
  1816. unsigned bytesOut = client? client->bytesOut() : 0;
  1817. unsigned elapsed = msTick() - qstart;
  1818. sink->noteQuery(msgctx.get(), peerStr, failed, bytesOut, elapsed, memused, slavesReplyLen, continuationNeeded);
  1819. if (continuationNeeded)
  1820. {
  1821. rawText.clear();
  1822. goto readAnother;
  1823. }
  1824. else
  1825. {
  1826. try
  1827. {
  1828. if (client && !isHTTP && !isStatus)
  1829. {
  1830. if (msgctx->getIntercept())
  1831. {
  1832. FlushingStringBuffer response(client, (protocolFlags & HPCC_PROTOCOL_BLOCKED), mlResponseFmt, (protocolFlags & HPCC_PROTOCOL_NATIVE_RAW), false, logctx);
  1833. response.startDataset("Tracing", NULL, (unsigned) -1);
  1834. msgctx->outputLogXML(response);
  1835. }
  1836. unsigned replyLen = 0;
  1837. client->write(&replyLen, sizeof(replyLen));
  1838. }
  1839. client.clear();
  1840. }
  1841. catch (IException * E)
  1842. {
  1843. StringBuffer error("RoxieSocketWorker failed to close socket ");
  1844. E->errorMessage(error);
  1845. logctx.CTXLOG("%s", error.str()); // MORE - audience?
  1846. E->Release();
  1847. }
  1848. catch(...)
  1849. {
  1850. logctx.CTXLOG("RoxieSocketWorker failed to close socket (Unknown exception)"); // MORE - audience?
  1851. }
  1852. }
  1853. }
  1854. };
  1855. IPooledThread *ProtocolSocketListener::createNew()
  1856. {
  1857. return new RoxieSocketWorker(this, ep);
  1858. }
  1859. void ProtocolSocketListener::runOnce(const char *query)
  1860. {
  1861. Owned<RoxieSocketWorker> p = new RoxieSocketWorker(this, ep);
  1862. p->runOnce(query);
  1863. }
  1864. IHpccProtocolListener *createProtocolListener(const char *protocol, IHpccProtocolMsgSink *sink, unsigned port, unsigned listenQueue)
  1865. {
  1866. if (traceLevel)
  1867. DBGLOG("Creating Roxie socket listener, protocol %s, pool size %d, listen queue %d%s", protocol, sink->getPoolSize(), listenQueue, sink->getIsSuspended() ? " SUSPENDED":"");
  1868. return new ProtocolSocketListener(sink, port, listenQueue);
  1869. }
  1870. extern IHpccProtocolPlugin *loadHpccProtocolPlugin(IHpccProtocolPluginContext *ctx, IActiveQueryLimiterFactory *_limiterFactory)
  1871. {
  1872. if (!queryLimiterFactory)
  1873. queryLimiterFactory.set(_limiterFactory);
  1874. if (global)
  1875. return global.getLink();
  1876. if (!ctx)
  1877. return NULL;
  1878. global.setown(new CHpccProtocolPlugin(*ctx));
  1879. return global.getLink();
  1880. }
  1881. //================================================================================================================================