redis.cpp 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2015 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 "jthread.hpp"
  15. #include "eclrtl.hpp"
  16. #include "jstring.hpp"
  17. #include "redis.hpp"
  18. #include "hiredis/hiredis.h"
  19. #define REDIS_VERSION "redis plugin 1.0.0"
  20. ECL_REDIS_API bool getECLPluginDefinition(ECLPluginDefinitionBlock *pb)
  21. {
  22. if (pb->size != sizeof(ECLPluginDefinitionBlock))
  23. return false;
  24. pb->magicVersion = PLUGIN_VERSION;
  25. pb->version = REDIS_VERSION;
  26. pb->moduleName = "lib_redis";
  27. pb->ECL = NULL;
  28. pb->flags = PLUGIN_IMPLICIT_MODULE;
  29. pb->description = "ECL plugin library for the C API hiredis\n";
  30. return true;
  31. }
  32. namespace RedisPlugin {
  33. class Connection;
  34. static const char * REDIS_LOCK_PREFIX = "redis_ecl_lock";
  35. static __thread Connection * cachedConnection;
  36. static __thread ThreadTermFunc threadHookChain;
  37. static void * allocateAndCopy(const void * src, size_t size)
  38. {
  39. return memcpy(rtlMalloc(size), src, size);
  40. }
  41. static StringBuffer & appendExpire(StringBuffer & buffer, unsigned expire)
  42. {
  43. if (expire > 0)
  44. buffer.append(" EX ").append(expire/1000);
  45. return buffer;
  46. }
  47. class Reply : public CInterface
  48. {
  49. public :
  50. inline Reply() : reply(NULL) { };
  51. inline Reply(void * _reply) : reply((redisReply*)_reply) { }
  52. inline Reply(redisReply * _reply) : reply(_reply) { }
  53. inline ~Reply()
  54. {
  55. if (reply)
  56. freeReplyObject(reply);
  57. }
  58. static Reply * createReply(void * _reply) { return new Reply(_reply); }
  59. inline const redisReply * query() const { return reply; }
  60. void setClear(redisReply * _reply)
  61. {
  62. if (reply)
  63. freeReplyObject(reply);
  64. reply = _reply;
  65. }
  66. private :
  67. redisReply * reply;
  68. };
  69. typedef Owned<RedisPlugin::Reply> OwnedReply;
  70. class Connection : public CInterface
  71. {
  72. public :
  73. Connection(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 _timeout);
  74. Connection(ICodeContext * ctx, const char * _options, const char * _ip, int _port, unsigned _serverIpPortPasswordHash, unsigned __int64 _database, const char * password, unsigned __int64 _timeout);
  75. ~Connection()
  76. {
  77. if (context)
  78. redisFree(context);
  79. }
  80. static Connection * createConnection(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 _timeout);
  81. //set
  82. template <class type> void set(ICodeContext * ctx, const char * key, type value, unsigned expire);
  83. template <class type> void set(ICodeContext * ctx, const char * key, size32_t valueSize, const type * value, unsigned expire);
  84. //get
  85. template <class type> void get(ICodeContext * ctx, const char * key, type & value);
  86. template <class type> void get(ICodeContext * ctx, const char * key, size_t & valueSize, type * & value);
  87. //-------------------------------LOCKING------------------------------------------------
  88. void lockSet(ICodeContext * ctx, const char * key, size32_t valueSize, const char * value, unsigned expire);
  89. void lockGet(ICodeContext * ctx, const char * key, size_t & valueSize, char * & value, const char * password);
  90. void unlock(ICodeContext * ctx, const char * key);
  91. //--------------------------------------------------------------------------------------
  92. void persist(ICodeContext * ctx, const char * key);
  93. void expire(ICodeContext * ctx, const char * key, unsigned _expire);
  94. void del(ICodeContext * ctx, const char * key);
  95. void clear(ICodeContext * ctx);
  96. unsigned __int64 dbSize(ICodeContext * ctx);
  97. bool exists(ICodeContext * ctx, const char * key);
  98. protected :
  99. void parseOptions(ICodeContext * ctx, const char * _options);
  100. void connect(ICodeContext * ctx, unsigned __int64 _database, const char * password);
  101. void selectDB(ICodeContext * ctx, unsigned __int64 _database);
  102. void resetContextErr();
  103. void readReply(Reply * reply);
  104. void readReplyAndAssert(Reply * reply, const char * msg);
  105. void readReplyAndAssertWithKey(Reply * reply, const char * msg, const char * key);
  106. void assertKey(const redisReply * reply, const char * key);
  107. void assertAuthorization(const redisReply * reply);
  108. void assertOnError(const redisReply * reply, const char * _msg);
  109. void assertOnCommandError(const redisReply * reply, const char * cmd);
  110. void assertOnCommandErrorWithDatabase(const redisReply * reply, const char * cmd);
  111. void assertOnCommandErrorWithKey(const redisReply * reply, const char * cmd, const char * key);
  112. void assertConnection();
  113. void updateTimeout(unsigned __int64 _timeout);
  114. static unsigned hashServerIpPortPassword(ICodeContext * ctx, const char * _options, const char * password);
  115. bool isSameConnection(ICodeContext * ctx, const char * _options, const char * password) const;
  116. //-------------------------------LOCKING------------------------------------------------
  117. void handleLockOnSet(ICodeContext * ctx, const char * key, const char * value, size_t size, unsigned expire);
  118. void handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAttr * retVal, const char * password);
  119. void encodeChannel(StringBuffer & channel, const char * key) const;
  120. bool lock(ICodeContext * ctx, const char * key, const char * channel);
  121. //--------------------------------------------------------------------------------------
  122. protected :
  123. redisContext * context;
  124. StringAttr options;
  125. StringAttr ip;
  126. unsigned serverIpPortPasswordHash;
  127. int port;
  128. unsigned __int64 timeout;
  129. unsigned __int64 database;
  130. };
  131. //The following class is here to ensure destruction of the cachedConnection within the main thread
  132. //as this is not handled by the thread hook mechanism.
  133. static class MainThreadCachedConnection
  134. {
  135. public :
  136. MainThreadCachedConnection() { }
  137. ~MainThreadCachedConnection()
  138. {
  139. if (cachedConnection)
  140. cachedConnection->Release();
  141. }
  142. } mainThread;
  143. static void releaseContext()
  144. {
  145. if (cachedConnection)
  146. {
  147. cachedConnection->Release();
  148. cachedConnection = NULL;
  149. }
  150. if (threadHookChain)
  151. {
  152. (*threadHookChain)();
  153. threadHookChain = NULL;
  154. }
  155. }
  156. Connection::Connection(ICodeContext * ctx, const char * _options, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
  157. : database(0), timeout(_timeout), port(0), serverIpPortPasswordHash(hashServerIpPortPassword(ctx, _options, password))
  158. {
  159. options.set(_options, strlen(_options));
  160. parseOptions(ctx, _options);
  161. connect(ctx, _database, password);
  162. }
  163. Connection::Connection(ICodeContext * ctx, const char * _options, const char * _ip, int _port, unsigned _serverIpPortPasswordHash, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
  164. : database(0), timeout(_timeout), serverIpPortPasswordHash(_serverIpPortPasswordHash), port(_port)
  165. {
  166. options.set(_options, strlen(_options));
  167. ip.set(_ip, strlen(_ip));
  168. connect(ctx, _database, password);
  169. }
  170. void Connection::connect(ICodeContext * ctx, unsigned __int64 _database, const char * password)
  171. {
  172. struct timeval to = { timeout/1000, (timeout%1000)*1000 };
  173. context = redisConnectWithTimeout(ip.str(), port, to);
  174. redisSetTimeout(context, to);
  175. assertConnection();
  176. //The following is the dissemination of the two methods authenticate(ctx, password) & selectDB(ctx, _database)
  177. //such that they may be pipelined to save an extra round trip to the server and back.
  178. if (password && *password)
  179. redisAppendCommand(context, "AUTH %b", password, strlen(password));
  180. if (database != _database)
  181. {
  182. VStringBuffer cmd("SELECT %" I64F "u", _database);
  183. redisAppendCommand(context, cmd.str());
  184. }
  185. //Now read replies.
  186. OwnedReply reply = new Reply();
  187. if (password && *password)
  188. readReplyAndAssert(reply, "server authentication failed");
  189. if (database != _database)
  190. {
  191. readReplyAndAssert(reply, "request to SELECT database failed");
  192. database = _database;
  193. }
  194. }
  195. bool Connection::isSameConnection(ICodeContext * ctx, const char * _options, const char * password) const
  196. {
  197. return (hashServerIpPortPassword(ctx, _options, password) == serverIpPortPasswordHash);
  198. }
  199. unsigned Connection::hashServerIpPortPassword(ICodeContext * ctx, const char * _options, const char * password)
  200. {
  201. return hashc((const unsigned char*)_options, strlen(_options), hashc((const unsigned char*)password, strlen(password), 0));
  202. }
  203. void Connection::parseOptions(ICodeContext * ctx, const char * _options)
  204. {
  205. StringArray optionStrings;
  206. optionStrings.appendList(_options, " ");
  207. ForEachItemIn(idx, optionStrings)
  208. {
  209. const char *opt = optionStrings.item(idx);
  210. if (strncmp(opt, "--SERVER=", 9) == 0)
  211. {
  212. opt += 9;
  213. StringArray splitPort;
  214. splitPort.appendList(opt, ":");
  215. if (splitPort.ordinality()==2)
  216. {
  217. ip.set(splitPort.item(0));
  218. port = atoi(splitPort.item(1));
  219. }
  220. }
  221. else
  222. {
  223. VStringBuffer err("RedisPlugin: unsupported option string %s", opt);
  224. rtlFail(0, err.str());
  225. }
  226. }
  227. if (ip.isEmpty())
  228. {
  229. ip.set("localhost");
  230. port = 6379;
  231. if (ctx)
  232. {
  233. VStringBuffer msg("Redis Plugin: WARNING - using default server (%s:%d)", ip.str(), port);
  234. ctx->logString(msg.str());
  235. }
  236. }
  237. }
  238. void Connection::resetContextErr()
  239. {
  240. if (context)
  241. context->err = REDIS_OK;
  242. }
  243. void Connection::readReply(Reply * reply)
  244. {
  245. redisReply * nakedReply = NULL;
  246. redisGetReply(context, (void**)&nakedReply);
  247. assertex(reply);
  248. reply->setClear(nakedReply);
  249. }
  250. void Connection::readReplyAndAssert(Reply * reply, const char * msg)
  251. {
  252. readReply(reply);
  253. assertex(reply);
  254. assertOnError(reply->query(), msg);
  255. }
  256. void Connection::readReplyAndAssertWithKey(Reply * reply, const char * msg, const char * key)
  257. {
  258. readReply(reply);
  259. assertex(reply);
  260. assertOnCommandErrorWithKey(reply->query(), msg, key);
  261. }
  262. Connection * Connection::createConnection(ICodeContext * ctx, const char * options, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
  263. {
  264. if (!cachedConnection)
  265. {
  266. cachedConnection = new Connection(ctx, options, _database, password, _timeout);
  267. threadHookChain = addThreadTermFunc(releaseContext);
  268. return LINK(cachedConnection);
  269. }
  270. if (cachedConnection->isSameConnection(ctx, options, password))
  271. {
  272. //MORE: should perhaps check that the connection has not expired (think hiredis REDIS_KEEPALIVE_INTERVAL is defaulted to 15s).
  273. //At present updateTimeout calls assertConnection.
  274. cachedConnection->resetContextErr();//reset the context err to allow reuse when an error previously occurred.
  275. cachedConnection->updateTimeout(_timeout);
  276. cachedConnection->selectDB(ctx, _database);
  277. return LINK(cachedConnection);
  278. }
  279. cachedConnection->Release();
  280. cachedConnection = NULL;
  281. cachedConnection = new Connection(ctx, options, _database, password, _timeout);
  282. return LINK(cachedConnection);
  283. }
  284. void Connection::selectDB(ICodeContext * ctx, unsigned __int64 _database)
  285. {
  286. if (database == _database)
  287. return;
  288. database = _database;
  289. VStringBuffer cmd("SELECT %" I64F "u", database);
  290. OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str()));
  291. assertOnCommandError(reply->query(), "SELECT");
  292. }
  293. void Connection::updateTimeout(unsigned __int64 _timeout)
  294. {
  295. if (timeout == _timeout)
  296. return;
  297. assertConnection();
  298. timeout = _timeout;
  299. struct timeval to = { timeout/1000, (timeout%1000)*1000 };
  300. assertex(context);
  301. if (redisSetTimeout(context, to) != REDIS_OK)
  302. {
  303. if (context->err)
  304. {
  305. VStringBuffer msg("RedisPlugin: failed to set timeout - %s", context->errstr);
  306. rtlFail(0, msg.str());
  307. }
  308. else
  309. rtlFail(0, "RedisPlugin: failed to set timeout - no message available");
  310. }
  311. }
  312. void Connection::assertOnError(const redisReply * reply, const char * _msg)
  313. {
  314. if (!reply)//MORE: should this be assertex(reply) instead?
  315. {
  316. //There should always be a context error if no reply error
  317. assertConnection();
  318. VStringBuffer msg("Redis Plugin: %s - %s", _msg, "neither 'reply' nor connection error available");
  319. rtlFail(0, msg.str());
  320. }
  321. else if (reply->type == REDIS_REPLY_ERROR)
  322. {
  323. assertAuthorization(reply);
  324. VStringBuffer msg("Redis Plugin: %s - %s", _msg, reply->str);
  325. rtlFail(0, msg.str());
  326. }
  327. }
  328. void Connection::assertOnCommandErrorWithKey(const redisReply * reply, const char * cmd, const char * key)
  329. {
  330. if (!reply)//MORE: should this be assertex(reply) instead?
  331. {
  332. //There should always be a context error if no reply error
  333. assertConnection();
  334. VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %" I64F "u failed with neither 'reply' nor connection error available", cmd, key, database);
  335. rtlFail(0, msg.str());
  336. }
  337. else if (reply->type == REDIS_REPLY_ERROR)
  338. {
  339. assertAuthorization(reply);
  340. VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %" I64F "u failed : %s", cmd, key, database, reply->str);
  341. rtlFail(0, msg.str());
  342. }
  343. }
  344. void Connection::assertOnCommandErrorWithDatabase(const redisReply * reply, const char * cmd)
  345. {
  346. if (!reply)//assertex(reply)?
  347. {
  348. //There should always be a context error if no reply error
  349. assertConnection();
  350. VStringBuffer msg("Redis Plugin: ERROR - %s on database %" I64F "u failed with neither 'reply' nor connection error available", cmd, database);
  351. rtlFail(0, msg.str());
  352. }
  353. else if (reply->type == REDIS_REPLY_ERROR)
  354. {
  355. assertAuthorization(reply);
  356. VStringBuffer msg("Redis Plugin: ERROR - %s on database %" I64F "u failed : %s", cmd, database, reply->str);
  357. rtlFail(0, msg.str());
  358. }
  359. }
  360. void Connection::assertOnCommandError(const redisReply * reply, const char * cmd)
  361. {
  362. if (!reply)//assertex(reply)?
  363. {
  364. //There should always be a context error if no reply error
  365. assertConnection();
  366. VStringBuffer msg("Redis Plugin: ERROR - %s failed with neither 'reply' nor connection error available", cmd);
  367. rtlFail(0, msg.str());
  368. }
  369. else if (reply->type == REDIS_REPLY_ERROR)
  370. {
  371. assertAuthorization(reply);
  372. VStringBuffer msg("Redis Plugin: ERROR - %s failed : %s", cmd, reply->str);
  373. rtlFail(0, msg.str());
  374. }
  375. }
  376. void Connection::assertAuthorization(const redisReply * reply)
  377. {
  378. if (strncmp(reply->str, "NOAUTH", 6) == 0)
  379. {
  380. VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
  381. rtlFail(0, msg.str());
  382. }
  383. }
  384. void Connection::assertKey(const redisReply * reply, const char * key)
  385. {
  386. if (reply && reply->type == REDIS_REPLY_NIL)
  387. {
  388. VStringBuffer msg("Redis Plugin: ERROR - the requested key '%s' does not exist on database %" I64F "u", key, database);
  389. rtlFail(0, msg.str());
  390. }
  391. }
  392. void Connection::assertConnection()
  393. {
  394. if (!context)
  395. rtlFail(0, "Redis Plugin: 'redisConnect' failed - no error available.");
  396. else if (context->err)
  397. {
  398. VStringBuffer msg("Redis Plugin: Connection failed - %s for %s:%u", context->errstr, ip.str(), port);
  399. rtlFail(0, msg.str());
  400. }
  401. }
  402. void Connection::clear(ICodeContext * ctx)
  403. {
  404. //NOTE: flush is the actual cache flush/clear/delete and not an io buffer flush.
  405. OwnedReply reply = Reply::createReply(redisCommand(context, "FLUSHDB"));//NOTE: FLUSHDB deletes current database where as FLUSHALL deletes all dbs.
  406. //NOTE: documented as never failing, but in case
  407. assertOnCommandErrorWithDatabase(reply->query(), "FlushDB");
  408. }
  409. void Connection::del(ICodeContext * ctx, const char * key)
  410. {
  411. OwnedReply reply = Reply::createReply(redisCommand(context, "DEL %b", key, strlen(key)));
  412. assertOnCommandErrorWithKey(reply->query(), "Del", key);
  413. }
  414. void Connection::persist(ICodeContext * ctx, const char * key)
  415. {
  416. OwnedReply reply = Reply::createReply(redisCommand(context, "PERSIST %b", key, strlen(key)));
  417. assertOnCommandErrorWithKey(reply->query(), "Persist", key);
  418. }
  419. void Connection::expire(ICodeContext * ctx, const char * key, unsigned _expire)
  420. {
  421. OwnedReply reply = Reply::createReply(redisCommand(context, "EXPIRE %b %u", key, strlen(key), _expire/1000));
  422. assertOnCommandErrorWithKey(reply->query(), "Expire", key);
  423. }
  424. bool Connection::exists(ICodeContext * ctx, const char * key)
  425. {
  426. OwnedReply reply = Reply::createReply(redisCommand(context, "EXISTS %b", key, strlen(key)));
  427. assertOnCommandErrorWithKey(reply->query(), "Exists", key);
  428. return (reply->query()->integer != 0);
  429. }
  430. unsigned __int64 Connection::dbSize(ICodeContext * ctx)
  431. {
  432. OwnedReply reply = Reply::createReply(redisCommand(context, "DBSIZE"));
  433. assertOnCommandErrorWithDatabase(reply->query(), "DBSIZE");
  434. return reply->query()->integer;
  435. }
  436. //-------------------------------------------SET-----------------------------------------
  437. //--OUTER--
  438. template<class type> void SyncRSet(ICodeContext * ctx, const char * _options, const char * key, type value, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
  439. {
  440. Owned<Connection> master = Connection::createConnection(ctx, _options, database, password, _timeout);
  441. master->set(ctx, key, value, expire);
  442. }
  443. //Set pointer types
  444. template<class type> void SyncRSet(ICodeContext * ctx, const char * _options, const char * key, size32_t valueSize, const type * value, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
  445. {
  446. Owned<Connection> master = Connection::createConnection(ctx, _options, database, password, _timeout);
  447. master->set(ctx, key, valueSize, value, expire);
  448. }
  449. //--INNER--
  450. template<class type> void Connection::set(ICodeContext * ctx, const char * key, type value, unsigned expire)
  451. {
  452. const char * _value = reinterpret_cast<const char *>(&value);//Do this even for char * to prevent compiler complaining
  453. StringBuffer cmd("SET %b %b");
  454. appendExpire(cmd, expire);
  455. OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), _value, sizeof(type)));
  456. assertOnCommandErrorWithKey(reply->query(), "SET", key);
  457. }
  458. template<class type> void Connection::set(ICodeContext * ctx, const char * key, size32_t valueSize, const type * value, unsigned expire)
  459. {
  460. const char * _value = reinterpret_cast<const char *>(value);//Do this even for char * to prevent compiler complaining
  461. StringBuffer cmd("SET %b %b");
  462. appendExpire(cmd, expire);
  463. OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), _value, (size_t)valueSize));
  464. assertOnCommandErrorWithKey(reply->query(), "SET", key);
  465. }
  466. //-------------------------------------------GET-----------------------------------------
  467. //--OUTER--
  468. template<class type> void SyncRGet(ICodeContext * ctx, const char * options, const char * key, type & returnValue, unsigned __int64 database, const char * password, unsigned __int64 _timeout)
  469. {
  470. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, _timeout);
  471. master->get(ctx, key, returnValue);
  472. }
  473. template<class type> void SyncRGet(ICodeContext * ctx, const char * options, const char * key, size_t & returnSize, type * & returnValue, unsigned __int64 database, const char * password, unsigned __int64 _timeout)
  474. {
  475. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, _timeout);
  476. master->get(ctx, key, returnSize, returnValue);
  477. }
  478. //--INNER--
  479. template<class type> void Connection::get(ICodeContext * ctx, const char * key, type & returnValue)
  480. {
  481. OwnedReply reply = Reply::createReply(redisCommand(context, "GET %b", key, strlen(key)));
  482. assertOnError(reply->query(), "GET");
  483. assertKey(reply->query(), key);
  484. size_t returnSize = reply->query()->len;
  485. if (sizeof(type)!=returnSize)
  486. {
  487. VStringBuffer msg("RedisPlugin: ERROR - Requested type of different size (%uB) from that stored (%uB).", (unsigned)sizeof(type), (unsigned)returnSize);
  488. rtlFail(0, msg.str());
  489. }
  490. memcpy(&returnValue, reply->query()->str, returnSize);
  491. }
  492. template<class type> void Connection::get(ICodeContext * ctx, const char * key, size_t & returnSize, type * & returnValue)
  493. {
  494. OwnedReply reply = Reply::createReply(redisCommand(context, "GET %b", key, strlen(key)));
  495. assertOnError(reply->query(), "GET");
  496. assertKey(reply->query(), key);
  497. returnSize = reply->query()->len;
  498. returnValue = reinterpret_cast<type*>(allocateAndCopy(reply->query()->str, returnSize));
  499. }
  500. //--------------------------------------------------------------------------------
  501. // ECL SERVICE ENTRYPOINTS
  502. //--------------------------------------------------------------------------------
  503. ECL_REDIS_API void ECL_REDIS_CALL RClear(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  504. {
  505. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  506. master->clear(ctx);
  507. }
  508. ECL_REDIS_API bool ECL_REDIS_CALL RExist(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  509. {
  510. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  511. return master->exists(ctx, key);
  512. }
  513. ECL_REDIS_API void ECL_REDIS_CALL RDel(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  514. {
  515. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  516. master->del(ctx, key);
  517. }
  518. ECL_REDIS_API void ECL_REDIS_CALL RPersist(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  519. {
  520. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  521. master->persist(ctx, key);
  522. }
  523. ECL_REDIS_API void ECL_REDIS_CALL RExpire(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, unsigned _expire, const char * password, unsigned __int64 timeout)
  524. {
  525. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  526. master->expire(ctx, key, _expire);
  527. }
  528. ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL RDBSize(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  529. {
  530. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  531. return master->dbSize(ctx);
  532. }
  533. //-----------------------------------SET------------------------------------------
  534. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetStr(ICodeContext * ctx, const char * key, size32_t valueSize, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  535. {
  536. SyncRSet(ctx, options, key, valueSize, value, database, expire, password, timeout);
  537. }
  538. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUChar(ICodeContext * ctx, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  539. {
  540. SyncRSet(ctx, options, key, (valueLength)*sizeof(UChar), value, database, expire, password, timeout);
  541. }
  542. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetInt(ICodeContext * ctx, const char * key, signed __int64 value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  543. {
  544. SyncRSet(ctx, options, key, value, database, expire, password, timeout);
  545. }
  546. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUInt(ICodeContext * ctx, const char * key, unsigned __int64 value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  547. {
  548. SyncRSet(ctx, options, key, value, database, expire, password, timeout);
  549. }
  550. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetReal(ICodeContext * ctx, const char * key, double value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  551. {
  552. SyncRSet(ctx, options, key, value, database, expire, password, timeout);
  553. }
  554. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetBool(ICodeContext * ctx, const char * key, bool value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  555. {
  556. SyncRSet(ctx, options, key, value, database, expire, password, timeout);
  557. }
  558. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetData(ICodeContext * ctx, const char * key, size32_t valueSize, const void * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  559. {
  560. SyncRSet(ctx, options, key, valueSize, value, database, expire, password, timeout);
  561. }
  562. ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUtf8(ICodeContext * ctx, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  563. {
  564. SyncRSet(ctx, options, key, rtlUtf8Size(valueLength, value), value, database, expire, password, timeout);
  565. }
  566. //-------------------------------------GET----------------------------------------
  567. ECL_REDIS_API bool ECL_REDIS_CALL SyncRGetBool(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  568. {
  569. bool value;
  570. SyncRGet(ctx, options, key, value, database, password, timeout);
  571. return value;
  572. }
  573. ECL_REDIS_API double ECL_REDIS_CALL SyncRGetDouble(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  574. {
  575. double value;
  576. SyncRGet(ctx, options, key, value, database, password, timeout);
  577. return value;
  578. }
  579. ECL_REDIS_API signed __int64 ECL_REDIS_CALL SyncRGetInt8(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  580. {
  581. signed __int64 value;
  582. SyncRGet(ctx, options, key, value, database, password, timeout);
  583. return value;
  584. }
  585. ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL SyncRGetUint8(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  586. {
  587. unsigned __int64 value;
  588. SyncRGet(ctx, options, key, value, database, password, timeout);
  589. return value;
  590. }
  591. ECL_REDIS_API void ECL_REDIS_CALL SyncRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  592. {
  593. size_t _returnSize;
  594. SyncRGet(ctx, options, key, _returnSize, returnValue, database, password, timeout);
  595. returnSize = static_cast<size32_t>(_returnSize);
  596. }
  597. ECL_REDIS_API void ECL_REDIS_CALL SyncRGetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  598. {
  599. size_t returnSize;
  600. SyncRGet(ctx, options, key, returnSize, returnValue, database, password, timeout);
  601. returnLength = static_cast<size32_t>(returnSize/sizeof(UChar));
  602. }
  603. ECL_REDIS_API void ECL_REDIS_CALL SyncRGetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  604. {
  605. size_t returnSize;
  606. SyncRGet(ctx, options, key, returnSize, returnValue, database, password, timeout);
  607. returnLength = static_cast<size32_t>(rtlUtf8Length(returnSize, returnValue));
  608. }
  609. ECL_REDIS_API void ECL_REDIS_CALL SyncRGetData(ICodeContext * ctx, size32_t & returnSize, void * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  610. {
  611. size_t _returnSize;
  612. SyncRGet(ctx, options, key, _returnSize, returnValue, database, password, timeout);
  613. returnSize = static_cast<size32_t>(_returnSize);
  614. }
  615. //----------------------------------LOCK------------------------------------------
  616. //-----------------------------------SET-----------------------------------------
  617. //Set pointer types
  618. void SyncLockRSet(ICodeContext * ctx, const char * _options, const char * key, size32_t valueSize, const char * value, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
  619. {
  620. Owned<Connection> master = Connection::createConnection(ctx, _options, database, password, _timeout);
  621. master->lockSet(ctx, key, valueSize, value, expire);
  622. }
  623. //--INNER--
  624. void Connection::lockSet(ICodeContext * ctx, const char * key, size32_t valueSize, const char * value, unsigned expire)
  625. {
  626. const char * _value = reinterpret_cast<const char *>(value);//Do this even for char * to prevent compiler complaining
  627. handleLockOnSet(ctx, key, _value, (size_t)valueSize, expire);
  628. }
  629. //-------------------------------------------GET-----------------------------------------
  630. //--OUTER--
  631. void SyncLockRGet(ICodeContext * ctx, const char * options, const char * key, size_t & returnSize, char * & returnValue, unsigned __int64 database, const char * password, unsigned __int64 _timeout)
  632. {
  633. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, _timeout);
  634. master->lockGet(ctx, key, returnSize, returnValue, password);
  635. }
  636. //--INNER--
  637. void Connection::lockGet(ICodeContext * ctx, const char * key, size_t & returnSize, char * & returnValue, const char * password)
  638. {
  639. MemoryAttr retVal;
  640. handleLockOnGet(ctx, key, &retVal, password);
  641. returnSize = retVal.length();
  642. returnValue = reinterpret_cast<char*>(retVal.detach());
  643. }
  644. //---------------------------------------------------------------------------------------
  645. void Connection::encodeChannel(StringBuffer & channel, const char * key) const
  646. {
  647. channel.append(REDIS_LOCK_PREFIX).append("_").append(key).append("_").append(database);
  648. }
  649. bool Connection::lock(ICodeContext * ctx, const char * key, const char * channel)
  650. {
  651. StringBuffer cmd("SET %b %b NX EX ");
  652. cmd.append(timeout/1000);
  653. OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), channel, strlen(channel)));
  654. assertOnError(reply->query(), cmd.append(" of the key '").append(key).append("' failed"));
  655. return (reply->query()->type == REDIS_REPLY_STATUS && strcmp(reply->query()->str, "OK") == 0);
  656. }
  657. void Connection::unlock(ICodeContext * ctx, const char * key)
  658. {
  659. //WATCH key, if altered between WATCH and EXEC abort all commands inbetween
  660. redisAppendCommand(context, "WATCH %b", key, strlen(key));
  661. redisAppendCommand(context, "GET %b", key, strlen(key));
  662. //Read replies
  663. OwnedReply reply = new Reply();
  664. readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//WATCH reply
  665. readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//GET reply
  666. //check if locked
  667. if (strncmp(reply->query()->str, REDIS_LOCK_PREFIX, strlen(REDIS_LOCK_PREFIX)) == 0)
  668. {
  669. //MULTI - all commands between MULTI and EXEC are considered an atomic transaction on the server
  670. redisAppendCommand(context, "MULTI");//MULTI
  671. redisAppendCommand(context, "DEL %b", key, strlen(key));//DEL
  672. redisAppendCommand(context, "EXEC");//EXEC
  673. #if(0)//Quick draw! You have 10s to manually send (via redis-cli) "set testlock foobar". The second myRedis.Exists('testlock') in redislockingtest.ecl should now return TRUE.
  674. sleep(10);
  675. #endif
  676. readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//MULTI reply
  677. readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//DEL reply
  678. readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//EXEC reply
  679. }
  680. //If the above is aborted, let the lock expire.
  681. }
  682. void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAttr * retVal, const char * password)
  683. {
  684. StringBuffer channel;
  685. encodeChannel(channel, key);
  686. //Query key and set lock if non existent
  687. if (lock(ctx, key, channel.str()))
  688. return;
  689. //SUB before GET
  690. //Requires separate connection from GET so that the replies are not mangled. This could be averted
  691. Owned<Connection> subConnection = new Connection(ctx, options.str(), ip.str(), port, serverIpPortPasswordHash, database, password, timeout);
  692. OwnedReply reply = Reply::createReply(redisCommand(subConnection->context, "SUBSCRIBE %b", channel.str(), channel.length()));
  693. assertOnCommandErrorWithKey(reply->query(), "GET", key);
  694. if (reply->query()->type == REDIS_REPLY_ARRAY && strcmp("subscribe", reply->query()->element[0]->str) != 0 )
  695. {
  696. VStringBuffer msg("Redis Plugin: ERROR - GET '%s' on database %" I64F "u failed : failed to register SUB", key, database);
  697. rtlFail(0, msg.str());
  698. }
  699. #if(0)
  700. {
  701. OwnedReply pubReply = Reply::createReply(redisCommand(context, "PUBLISH %b %b", channel.str(), channel.length(), "foo", 3));
  702. assertOnError(pubReply->query(), "pub fail");
  703. }
  704. #endif
  705. //Now GET
  706. reply->setClear((redisReply*)redisCommand(context, "GET %b", key, strlen(key)));
  707. assertOnCommandErrorWithKey(reply->query(), "GET", key);
  708. assertKey(reply->query(), key);
  709. #if(0)
  710. {
  711. OwnedReply pubReply = Reply::createReply(redisCommand(context, "PUBLISH %b %b", channel.str(), channel.length(), "foo", 3));
  712. assertOnError(pubReply->query(), "pub fail");
  713. }
  714. #endif
  715. //Check if returned value is locked
  716. if (strncmp(reply->query()->str, REDIS_LOCK_PREFIX, strlen(REDIS_LOCK_PREFIX)) != 0)
  717. {
  718. //Not locked so return value
  719. retVal->set(reply->query()->len, reply->query()->str);
  720. return;
  721. }
  722. else
  723. {
  724. //Check that we SUBSCRIBEd to the correct channel (which could have been manually SET).
  725. if (strcmp(reply->query()->str, channel.str()) !=0 )
  726. {
  727. VStringBuffer msg("Redis Plugin: ERROR - the key '%s', on database %" I64F "u, is locked with a channel ('%s') different to that subscribed to (%s).", key, database, reply->query()->str, channel.str());
  728. rtlFail(0, msg.str());
  729. //MORE: We could attempt to recover at this stage by subscribing to the channel that the key was actually locked with.
  730. //However, we may have missed the massage publication already or by then.
  731. //If we ever changed the semantics of the 'timeout' to be that of these plugin functions rather than each redis call, we might as well
  732. //subscribe again if there was time left on the clock.
  733. //Since they are not, we could, though is this desirable behaviour?
  734. }
  735. #if(0)//Added to allow for manual pub testing via redis-cli
  736. struct timeval to = { 10, 0 };//10secs
  737. redisSetTimeout(subConnection->context, to);
  738. #endif
  739. //Locked so SUBSCRIBE
  740. redisReply * nakedReply = NULL;
  741. bool err = redisGetReply(subConnection->context, (void**)&nakedReply);
  742. reply->setClear(nakedReply);
  743. if (err != REDIS_OK)
  744. rtlFail(0, "RedisPlugin: ERROR - GET timed out.");
  745. assertOnCommandErrorWithKey(nakedReply, "GET", key);
  746. if (nakedReply->type == REDIS_REPLY_ARRAY && strcmp("message", nakedReply->element[0]->str) == 0)
  747. {
  748. retVal->set(nakedReply->element[2]->len, nakedReply->element[2]->str);//return the published value rather than another (WATCHed) GET.
  749. return;
  750. }
  751. }
  752. throwUnexpected();
  753. }
  754. void Connection::handleLockOnSet(ICodeContext * ctx, const char * key, const char * value, size_t size, unsigned expire)
  755. {
  756. StringBuffer cmd("SET %b %b");
  757. RedisPlugin::appendExpire(cmd, expire);
  758. //Due to locking logic surfacing into ECL, any locking.set (such as this is) assumes that they own the lock and therefore go ahead and set regardless.
  759. //It is possible for a process/call to 'own' a lock and store this info in the LockObject, however, this prevents sharing between clients.
  760. redisAppendCommand(context, cmd.str(), key, strlen(key), value, size);//SET
  761. StringBuffer channel;
  762. encodeChannel(channel, key);
  763. redisAppendCommand(context, "PUBLISH %b %b", channel.str(), channel.length(), value, size);//PUB
  764. //Now read and assert replies
  765. OwnedReply replyContainer = new Reply();
  766. readReplyAndAssertWithKey(replyContainer, "SET", key);//SET reply
  767. readReplyAndAssertWithKey(replyContainer, "PUB for the key", key);//PUB reply
  768. //NOTE: Pipelining the above commands may not be the desired behaviour but instead only PUBLISH upon a successful SET. Doing both regardless, does however ensure
  769. //(assuming only the SET fails) that any subscribers do in fact get their requested key-value even if the SET fails. However, this may not be expected behaviour
  770. //as it's now possible for the key-value to actually exists in the cache when it was retrieved via redis plugin get function. This is documented in the README.
  771. //Further more, it is possible that the locked value and thus the channel stored within the key is not that expected, i.e. computed via encodeChannel() (e.g.
  772. //if set by a non-conforming external client/process). It is however, possible to account for this via using a GETSET instead of just the SET. This returns the old
  773. //value stored, this can then be checked if it is a lock (i.e. has at least the "redis_key_lock prefix"), if it doesn't, PUB on the channel from encodeChannel(),
  774. //otherwise PUB on the value retrieved from GETSET or possibly only if it at least has the prefix "redis_key_lock".
  775. //This would however, prevent the two commands from being pipelined, as the GETSET would need to return before publishing.
  776. }
  777. //--------------------------------------------------------------------------------
  778. // ECL SERVICE ENTRYPOINTS
  779. //--------------------------------------------------------------------------------
  780. //-----------------------------------SET------------------------------------------
  781. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetStr(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  782. {
  783. SyncLockRSet(ctx, options, key, valueLength, value, expire, database, password, timeout);
  784. returnLength = valueLength;
  785. returnValue = (char*)allocateAndCopy(value, valueLength);
  786. }
  787. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  788. {
  789. unsigned valueSize = (valueLength)*sizeof(UChar);
  790. SyncLockRSet(ctx, options, key, valueSize, (char*)value, expire, database, password, timeout);
  791. returnLength= valueLength;
  792. returnValue = (UChar*)allocateAndCopy(value, valueSize);
  793. }
  794. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
  795. {
  796. unsigned valueSize = rtlUtf8Size(valueLength, value);
  797. SyncLockRSet(ctx, options, key, valueSize, value, expire, database, password, timeout);
  798. returnLength = valueLength;
  799. returnValue = (char*)allocateAndCopy(value, valueSize);
  800. }
  801. //-------------------------------------GET----------------------------------------
  802. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  803. {
  804. size_t _returnSize;
  805. SyncLockRGet(ctx, options, key, _returnSize, returnValue, database, password, timeout);
  806. returnSize = static_cast<size32_t>(_returnSize);
  807. }
  808. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  809. {
  810. size_t returnSize;
  811. char * _returnValue;
  812. SyncLockRGet(ctx, options, key, returnSize, _returnValue, database, password, timeout);
  813. returnValue = (UChar*)_returnValue;
  814. returnLength = static_cast<size32_t>(returnSize/sizeof(UChar));
  815. }
  816. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  817. {
  818. size_t returnSize;
  819. SyncLockRGet(ctx, options, key, returnSize, returnValue, database, password, timeout);
  820. returnLength = static_cast<size32_t>(rtlUtf8Length(returnSize, returnValue));
  821. }
  822. ECL_REDIS_API void ECL_REDIS_CALL SyncLockRUnlock(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
  823. {
  824. Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
  825. master->unlock(ctx, key);
  826. }
  827. }//close namespace