hircluster.c 118 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036403740384039404040414042404340444045404640474048404940504051405240534054405540564057405840594060406140624063406440654066406740684069407040714072407340744075407640774078407940804081408240834084408540864087408840894090409140924093409440954096409740984099410041014102410341044105410641074108410941104111411241134114411541164117411841194120412141224123412441254126412741284129413041314132413341344135413641374138413941404141414241434144414541464147414841494150415141524153415441554156415741584159416041614162416341644165416641674168416941704171417241734174417541764177417841794180418141824183418441854186418741884189419041914192419341944195419641974198419942004201420242034204420542064207420842094210421142124213421442154216421742184219422042214222422342244225422642274228422942304231423242334234423542364237423842394240424142424243424442454246424742484249425042514252425342544255425642574258425942604261426242634264426542664267426842694270427142724273427442754276427742784279428042814282428342844285428642874288428942904291429242934294429542964297429842994300430143024303430443054306430743084309431043114312431343144315431643174318431943204321432243234324432543264327432843294330433143324333433443354336433743384339434043414342434343444345434643474348434943504351435243534354435543564357435843594360436143624363436443654366436743684369437043714372437343744375437643774378437943804381438243834384438543864387438843894390439143924393439443954396439743984399440044014402440344044405440644074408440944104411441244134414441544164417441844194420442144224423442444254426442744284429443044314432443344344435443644374438443944404441444244434444444544464447444844494450445144524453445444554456445744584459446044614462446344644465446644674468446944704471447244734474447544764477447844794480448144824483448444854486448744884489449044914492449344944495449644974498449945004501450245034504450545064507450845094510451145124513451445154516451745184519452045214522452345244525452645274528452945304531453245334534453545364537453845394540454145424543454445454546454745484549455045514552455345544555455645574558455945604561456245634564456545664567456845694570457145724573457445754576457745784579458045814582458345844585458645874588458945904591459245934594459545964597459845994600460146024603460446054606460746084609461046114612461346144615461646174618461946204621462246234624462546264627462846294630463146324633463446354636463746384639464046414642464346444645464646474648464946504651465246534654465546564657465846594660466146624663466446654666466746684669467046714672467346744675467646774678467946804681468246834684468546864687468846894690469146924693469446954696469746984699470047014702470347044705470647074708470947104711471247134714471547164717471847194720472147224723472447254726472747284729473047314732473347344735473647374738473947404741474247434744474547464747
  1. #include "fmacros.h"
  2. #include <stdio.h>
  3. #include <string.h>
  4. #include <stdlib.h>
  5. #include <errno.h>
  6. #include <ctype.h>
  7. #include "hircluster.h"
  8. #include "hiutil.h"
  9. #include "adlist.h"
  10. #include "hiarray.h"
  11. #include "command.h"
  12. #include "dict.c"
  13. #define REDIS_COMMAND_CLUSTER_NODES "CLUSTER NODES"
  14. #define REDIS_COMMAND_CLUSTER_SLOTS "CLUSTER SLOTS"
  15. #define REDIS_COMMAND_ASKING "ASKING"
  16. #define REDIS_COMMAND_PING "PING"
  17. #define REDIS_PROTOCOL_ASKING "*1\r\n$6\r\nASKING\r\n"
  18. #define IP_PORT_SEPARATOR ":"
  19. #define CLUSTER_ADDRESS_SEPARATOR ","
  20. #define CLUSTER_DEFAULT_MAX_REDIRECT_COUNT 5
  21. typedef struct cluster_async_data
  22. {
  23. redisClusterAsyncContext *acc;
  24. struct cmd *command;
  25. redisClusterCallbackFn *callback;
  26. int retry_count;
  27. void *privdata;
  28. }cluster_async_data;
  29. typedef enum CLUSTER_ERR_TYPE{
  30. CLUSTER_NOT_ERR = 0,
  31. CLUSTER_ERR_MOVED,
  32. CLUSTER_ERR_ASK,
  33. CLUSTER_ERR_TRYAGAIN,
  34. CLUSTER_ERR_CROSSSLOT,
  35. CLUSTER_ERR_CLUSTERDOWN,
  36. CLUSTER_ERR_SENTINEL
  37. }CLUSTER_ERR_TYPE;
  38. static void cluster_node_deinit(cluster_node *node);
  39. static void cluster_slot_destroy(cluster_slot *slot);
  40. static void cluster_open_slot_destroy(copen_slot *oslot);
  41. void listClusterNodeDestructor(void *val)
  42. {
  43. cluster_node_deinit(val);
  44. hi_free(val);
  45. }
  46. void listClusterSlotDestructor(void *val)
  47. {
  48. cluster_slot_destroy(val);
  49. }
  50. unsigned int dictSdsHash(const void *key) {
  51. return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
  52. }
  53. int dictSdsKeyCompare(void *privdata, const void *key1,
  54. const void *key2)
  55. {
  56. int l1,l2;
  57. DICT_NOTUSED(privdata);
  58. l1 = sdslen((sds)key1);
  59. l2 = sdslen((sds)key2);
  60. if (l1 != l2) return 0;
  61. return memcmp(key1, key2, l1) == 0;
  62. }
  63. void dictSdsDestructor(void *privdata, void *val)
  64. {
  65. DICT_NOTUSED(privdata);
  66. sdsfree(val);
  67. }
  68. void dictClusterNodeDestructor(void *privdata, void *val)
  69. {
  70. DICT_NOTUSED(privdata);
  71. cluster_node_deinit(val);
  72. hi_free(val);
  73. }
  74. /* Cluster nodes hash table, mapping nodes
  75. * name(437c719f50dc9d0745032f3b280ce7ecc40792ac)
  76. * or addresses(1.2.3.4:6379) to clusterNode structures.
  77. * Those nodes need destroy.
  78. */
  79. dictType clusterNodesDictType = {
  80. dictSdsHash, /* hash function */
  81. NULL, /* key dup */
  82. NULL, /* val dup */
  83. dictSdsKeyCompare, /* key compare */
  84. dictSdsDestructor, /* key destructor */
  85. dictClusterNodeDestructor /* val destructor */
  86. };
  87. /* Cluster nodes hash table, mapping nodes
  88. * name(437c719f50dc9d0745032f3b280ce7ecc40792ac)
  89. * or addresses(1.2.3.4:6379) to clusterNode structures.
  90. * Those nodes do not need destroy.
  91. */
  92. dictType clusterNodesRefDictType = {
  93. dictSdsHash, /* hash function */
  94. NULL, /* key dup */
  95. NULL, /* val dup */
  96. dictSdsKeyCompare, /* key compare */
  97. dictSdsDestructor, /* key destructor */
  98. NULL /* val destructor */
  99. };
  100. void listCommandFree(void *command)
  101. {
  102. struct cmd *cmd = command;
  103. command_destroy(cmd);
  104. }
  105. /* Defined in hiredis.c */
  106. void __redisSetError(redisContext *c, int type, const char *str);
  107. /* Forward declaration of function in hiredis.c */
  108. int __redisAppendCommand(redisContext *c, const char *cmd, size_t len);
  109. /* Helper function for the redisClusterCommand* family of functions.
  110. *
  111. * Write a formatted command to the output buffer. If the given context is
  112. * blocking, immediately read the reply into the "reply" pointer. When the
  113. * context is non-blocking, the "reply" pointer will not be used and the
  114. * command is simply appended to the write buffer.
  115. *
  116. * Returns the reply when a reply was succesfully retrieved. Returns NULL
  117. * otherwise. When NULL is returned in a blocking context, the error field
  118. * in the context will be set.
  119. */
  120. static void *__redisBlockForReply(redisContext *c) {
  121. void *reply;
  122. if (c->flags & REDIS_BLOCK) {
  123. if (redisGetReply(c,&reply) != REDIS_OK)
  124. return NULL;
  125. return reply;
  126. }
  127. return NULL;
  128. }
  129. /* -----------------------------------------------------------------------------
  130. * Key space handling
  131. * -------------------------------------------------------------------------- */
  132. /* We have 16384 hash slots. The hash slot of a given key is obtained
  133. * as the least significant 14 bits of the crc16 of the key.
  134. *
  135. * However if the key contains the {...} pattern, only the part between
  136. * { and } is hashed. This may be useful in the future to force certain
  137. * keys to be in the same node (assuming no resharding is in progress). */
  138. static unsigned int keyHashSlot(char *key, int keylen) {
  139. int s, e; /* start-end indexes of { and } */
  140. for (s = 0; s < keylen; s++)
  141. if (key[s] == '{') break;
  142. /* No '{' ? Hash the whole key. This is the base case. */
  143. if (s == keylen) return crc16(key,keylen) & 0x3FFF;
  144. /* '{' found? Check if we have the corresponding '}'. */
  145. for (e = s+1; e < keylen; e++)
  146. if (key[e] == '}') break;
  147. /* No '}' or nothing betweeen {} ? Hash the whole key. */
  148. if (e == keylen || e == s+1) return crc16(key,keylen) & 0x3FFF;
  149. /* If we are here there is both a { and a } on its right. Hash
  150. * what is in the middle between { and }. */
  151. return crc16(key+s+1,e-s-1) & 0x3FFF;
  152. }
  153. static void __redisClusterSetError(redisClusterContext *cc, int type, const char *str) {
  154. size_t len;
  155. if(cc == NULL){
  156. return;
  157. }
  158. cc->err = type;
  159. if (str != NULL) {
  160. len = strlen(str);
  161. len = len < (sizeof(cc->errstr)-1) ? len : (sizeof(cc->errstr)-1);
  162. memcpy(cc->errstr,str,len);
  163. cc->errstr[len] = '\0';
  164. } else {
  165. /* Only REDIS_ERR_IO may lack a description! */
  166. assert(type == REDIS_ERR_IO);
  167. __redis_strerror_r(errno, cc->errstr, sizeof(cc->errstr));
  168. }
  169. }
  170. static int cluster_reply_error_type(redisReply *reply)
  171. {
  172. if(reply == NULL)
  173. {
  174. return REDIS_ERR;
  175. }
  176. if(reply->type == REDIS_REPLY_ERROR)
  177. {
  178. if((int)strlen(REDIS_ERROR_MOVED) < reply->len &&
  179. strncmp(reply->str, REDIS_ERROR_MOVED, strlen(REDIS_ERROR_MOVED)) == 0)
  180. {
  181. return CLUSTER_ERR_MOVED;
  182. }
  183. else if((int)strlen(REDIS_ERROR_ASK) < reply->len &&
  184. strncmp(reply->str, REDIS_ERROR_ASK, strlen(REDIS_ERROR_ASK)) == 0)
  185. {
  186. return CLUSTER_ERR_ASK;
  187. }
  188. else if((int)strlen(REDIS_ERROR_TRYAGAIN) < reply->len &&
  189. strncmp(reply->str, REDIS_ERROR_TRYAGAIN, strlen(REDIS_ERROR_TRYAGAIN)) == 0)
  190. {
  191. return CLUSTER_ERR_TRYAGAIN;
  192. }
  193. else if((int)strlen(REDIS_ERROR_CROSSSLOT) < reply->len &&
  194. strncmp(reply->str, REDIS_ERROR_CROSSSLOT, strlen(REDIS_ERROR_CROSSSLOT)) == 0)
  195. {
  196. return CLUSTER_ERR_CROSSSLOT;
  197. }
  198. else if((int)strlen(REDIS_ERROR_CLUSTERDOWN) < reply->len &&
  199. strncmp(reply->str, REDIS_ERROR_CLUSTERDOWN, strlen(REDIS_ERROR_CLUSTERDOWN)) == 0)
  200. {
  201. return CLUSTER_ERR_CLUSTERDOWN;
  202. }
  203. else
  204. {
  205. return CLUSTER_ERR_SENTINEL;
  206. }
  207. }
  208. return CLUSTER_NOT_ERR;
  209. }
  210. static int cluster_node_init(cluster_node *node)
  211. {
  212. if(node == NULL){
  213. return REDIS_ERR;
  214. }
  215. node->name = NULL;
  216. node->addr = NULL;
  217. node->host = NULL;
  218. node->port = 0;
  219. node->role = REDIS_ROLE_NULL;
  220. node->myself = 0;
  221. node->slaves = NULL;
  222. node->con = NULL;
  223. node->acon = NULL;
  224. node->slots = NULL;
  225. node->failure_count = 0;
  226. node->data = NULL;
  227. node->migrating = NULL;
  228. node->importing = NULL;
  229. return REDIS_OK;
  230. }
  231. static void cluster_node_deinit(cluster_node *node)
  232. {
  233. copen_slot **oslot;
  234. if(node == NULL)
  235. {
  236. return;
  237. }
  238. sdsfree(node->name);
  239. sdsfree(node->addr);
  240. sdsfree(node->host);
  241. node->port = 0;
  242. node->role = REDIS_ROLE_NULL;
  243. node->myself = 0;
  244. if(node->con != NULL)
  245. {
  246. redisFree(node->con);
  247. }
  248. if(node->acon != NULL)
  249. {
  250. redisAsyncFree(node->acon);
  251. }
  252. if(node->slots != NULL)
  253. {
  254. listRelease(node->slots);
  255. }
  256. if(node->slaves != NULL)
  257. {
  258. listRelease(node->slaves);
  259. }
  260. if(node->migrating)
  261. {
  262. while(hiarray_n(node->migrating))
  263. {
  264. oslot = hiarray_pop(node->migrating);
  265. cluster_open_slot_destroy(*oslot);
  266. }
  267. hiarray_destroy(node->migrating);
  268. node->migrating = NULL;
  269. }
  270. if(node->importing)
  271. {
  272. while(hiarray_n(node->importing))
  273. {
  274. oslot = hiarray_pop(node->importing);
  275. cluster_open_slot_destroy(*oslot);
  276. }
  277. hiarray_destroy(node->importing);
  278. node->importing = NULL;
  279. }
  280. }
  281. static int cluster_slot_init(cluster_slot *slot, cluster_node *node)
  282. {
  283. slot->start = 0;
  284. slot->end = 0;
  285. slot->node = node;
  286. return REDIS_OK;
  287. }
  288. static cluster_slot *cluster_slot_create(cluster_node *node)
  289. {
  290. cluster_slot *slot;
  291. slot = hi_alloc(sizeof(*slot));
  292. if(slot == NULL){
  293. return NULL;
  294. }
  295. cluster_slot_init(slot, node);
  296. if(node != NULL){
  297. ASSERT(node->role == REDIS_ROLE_MASTER);
  298. if(node->slots == NULL){
  299. node->slots = listCreate();
  300. if(node->slots == NULL)
  301. {
  302. cluster_slot_destroy(slot);
  303. return NULL;
  304. }
  305. node->slots->free = listClusterSlotDestructor;
  306. }
  307. listAddNodeTail(node->slots, slot);
  308. }
  309. return slot;
  310. }
  311. static int cluster_slot_ref_node(cluster_slot * slot, cluster_node *node)
  312. {
  313. if(slot == NULL || node == NULL){
  314. return REDIS_ERR;
  315. }
  316. if(node->role != REDIS_ROLE_MASTER){
  317. return REDIS_ERR;
  318. }
  319. if(node->slots == NULL){
  320. node->slots = listCreate();
  321. if(node->slots == NULL)
  322. {
  323. return REDIS_ERR;
  324. }
  325. node->slots->free = listClusterSlotDestructor;
  326. }
  327. listAddNodeTail(node->slots, slot);
  328. slot->node = node;
  329. return REDIS_OK;
  330. }
  331. static void cluster_slot_destroy(cluster_slot *slot)
  332. {
  333. slot->start = 0;
  334. slot->end = 0;
  335. slot->node = NULL;
  336. hi_free(slot);
  337. }
  338. static copen_slot *cluster_open_slot_create(uint32_t slot_num, int migrate,
  339. sds remote_name, cluster_node *node)
  340. {
  341. copen_slot *oslot;
  342. oslot = hi_alloc(sizeof(*oslot));
  343. if(oslot == NULL){
  344. return NULL;
  345. }
  346. oslot->slot_num = 0;
  347. oslot->migrate = 0;
  348. oslot->node = NULL;
  349. oslot->remote_name = NULL;
  350. oslot->slot_num = slot_num;
  351. oslot->migrate = migrate;
  352. oslot->node = node;
  353. oslot->remote_name = sdsdup(remote_name);
  354. return oslot;
  355. }
  356. static void cluster_open_slot_destroy(copen_slot *oslot)
  357. {
  358. oslot->slot_num = 0;
  359. oslot->migrate = 0;
  360. oslot->node = NULL;
  361. if(oslot->remote_name != NULL){
  362. sdsfree(oslot->remote_name);
  363. oslot->remote_name = NULL;
  364. }
  365. hi_free(oslot);
  366. }
  367. /**
  368. * Return a new node with the "cluster slots" command reply.
  369. */
  370. static cluster_node *node_get_with_slots(
  371. redisClusterContext *cc, redisReply *host_elem,
  372. redisReply *port_elem, uint8_t role)
  373. {
  374. cluster_node *node = NULL;
  375. if(host_elem == NULL || port_elem == NULL){
  376. return NULL;
  377. }
  378. if(host_elem->type != REDIS_REPLY_STRING ||
  379. host_elem->len <= 0){
  380. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  381. "Command(cluster slots) reply error: "
  382. "node ip is not string.");
  383. goto error;
  384. }
  385. if(port_elem->type != REDIS_REPLY_INTEGER ||
  386. port_elem->integer <= 0){
  387. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  388. "Command(cluster slots) reply error: "
  389. "node port is not integer.");
  390. goto error;
  391. }
  392. if(!hi_valid_port((int)port_elem->integer)){
  393. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  394. "Command(cluster slots) reply error: "
  395. "node port is not valid.");
  396. goto error;
  397. }
  398. node = hi_alloc(sizeof(cluster_node));
  399. if(node == NULL){
  400. __redisClusterSetError(cc,
  401. REDIS_ERR_OOM,"Out of memory");
  402. goto error;
  403. }
  404. cluster_node_init(node);
  405. if(role == REDIS_ROLE_MASTER){
  406. node->slots = listCreate();
  407. if(node->slots == NULL){
  408. hi_free(node);
  409. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  410. "slots for node listCreate error");
  411. goto error;
  412. }
  413. node->slots->free = listClusterSlotDestructor;
  414. }
  415. node->name = NULL;
  416. node->addr = sdsnewlen(host_elem->str, host_elem->len);
  417. node->addr = sdscatfmt(node->addr, ":%i", port_elem->integer);
  418. node->host = sdsnewlen(host_elem->str, host_elem->len);
  419. node->port = (int)port_elem->integer;
  420. node->role = role;
  421. return node;
  422. error:
  423. if(node != NULL){
  424. hi_free(node);
  425. }
  426. return NULL;
  427. }
  428. /**
  429. * Return a new node with the "cluster nodes" command reply.
  430. */
  431. static cluster_node *node_get_with_nodes(
  432. redisClusterContext *cc,
  433. sds *node_infos, int info_count, uint8_t role)
  434. {
  435. sds *ip_port = NULL;
  436. int count_ip_port = 0;
  437. cluster_node *node;
  438. if(info_count < 8)
  439. {
  440. return NULL;
  441. }
  442. node = hi_alloc(sizeof(cluster_node));
  443. if(node == NULL)
  444. {
  445. __redisClusterSetError(cc,
  446. REDIS_ERR_OOM,"Out of memory");
  447. goto error;
  448. }
  449. cluster_node_init(node);
  450. if(role == REDIS_ROLE_MASTER)
  451. {
  452. node->slots = listCreate();
  453. if(node->slots == NULL)
  454. {
  455. hi_free(node);
  456. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  457. "slots for node listCreate error");
  458. goto error;
  459. }
  460. node->slots->free = listClusterSlotDestructor;
  461. }
  462. node->name = node_infos[0];
  463. node->addr = node_infos[1];
  464. ip_port = sdssplitlen(node_infos[1], sdslen(node_infos[1]),
  465. IP_PORT_SEPARATOR, strlen(IP_PORT_SEPARATOR), &count_ip_port);
  466. if(ip_port == NULL || count_ip_port != 2)
  467. {
  468. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  469. "split ip port error");
  470. goto error;
  471. }
  472. node->host = ip_port[0];
  473. node->port = hi_atoi(ip_port[1], sdslen(ip_port[1]));
  474. node->role = role;
  475. sdsfree(ip_port[1]);
  476. free(ip_port);
  477. node_infos[0] = NULL;
  478. node_infos[1] = NULL;
  479. return node;
  480. error:
  481. if(ip_port != NULL)
  482. {
  483. sdsfreesplitres(ip_port, count_ip_port);
  484. }
  485. if(node != NULL)
  486. {
  487. hi_free(node);
  488. }
  489. return NULL;
  490. }
  491. static void cluster_nodes_swap_ctx(dict *nodes_f, dict *nodes_t)
  492. {
  493. dictIterator *di;
  494. dictEntry *de_f, *de_t;
  495. cluster_node *node_f, *node_t;
  496. redisContext *c;
  497. redisAsyncContext *ac;
  498. if(nodes_f == NULL || nodes_t == NULL){
  499. return;
  500. }
  501. di = dictGetIterator(nodes_t);
  502. while((de_t = dictNext(di)) != NULL){
  503. node_t = dictGetEntryVal(de_t);
  504. if(node_t == NULL){
  505. continue;
  506. }
  507. de_f = dictFind(nodes_f, node_t->addr);
  508. if(de_f == NULL){
  509. continue;
  510. }
  511. node_f = dictGetEntryVal(de_f);
  512. if(node_f->con != NULL){
  513. c = node_f->con;
  514. node_f->con = node_t->con;
  515. node_t->con = c;
  516. }
  517. if(node_f->acon != NULL){
  518. ac = node_f->acon;
  519. node_f->acon = node_t->acon;
  520. node_t->acon = ac;
  521. node_t->acon->data = node_t;
  522. if (node_f->acon)
  523. node_f->acon->data = node_f;
  524. }
  525. }
  526. dictReleaseIterator(di);
  527. }
  528. static int
  529. cluster_slot_start_cmp(const void *t1, const void *t2)
  530. {
  531. const cluster_slot **s1 = t1, **s2 = t2;
  532. return (*s1)->start > (*s2)->start?1:-1;
  533. }
  534. static int
  535. cluster_master_slave_mapping_with_name(redisClusterContext *cc,
  536. dict **nodes, cluster_node *node, sds master_name)
  537. {
  538. int ret;
  539. dictEntry *di;
  540. cluster_node *node_old;
  541. listNode *lnode;
  542. if(node == NULL || master_name == NULL)
  543. {
  544. return REDIS_ERR;
  545. }
  546. if(*nodes == NULL)
  547. {
  548. *nodes = dictCreate(
  549. &clusterNodesRefDictType, NULL);
  550. }
  551. di = dictFind(*nodes, master_name);
  552. if(di == NULL)
  553. {
  554. ret = dictAdd(*nodes,
  555. sdsnewlen(master_name, sdslen(master_name)), node);
  556. if(ret != DICT_OK)
  557. {
  558. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  559. "the address already exists in the nodes");
  560. return REDIS_ERR;
  561. }
  562. }
  563. else
  564. {
  565. node_old = dictGetEntryVal(di);
  566. if(node_old == NULL)
  567. {
  568. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  569. "dict get value null");
  570. return REDIS_ERR;
  571. }
  572. if(node->role == REDIS_ROLE_MASTER &&
  573. node_old->role == REDIS_ROLE_MASTER)
  574. {
  575. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  576. "two masters have the same name");
  577. return REDIS_ERR;
  578. }
  579. else if(node->role == REDIS_ROLE_MASTER
  580. && node_old->role == REDIS_ROLE_SLAVE)
  581. {
  582. if(node->slaves == NULL)
  583. {
  584. node->slaves = listCreate();
  585. if(node->slaves == NULL)
  586. {
  587. __redisClusterSetError(cc,REDIS_ERR_OOM,
  588. "Out of memory");
  589. return REDIS_ERR;
  590. }
  591. node->slaves->free =
  592. listClusterNodeDestructor;
  593. }
  594. if(node_old->slaves != NULL)
  595. {
  596. node_old->slaves->free = NULL;
  597. while(listLength(node_old->slaves) > 0)
  598. {
  599. lnode = listFirst(node_old->slaves);
  600. listAddNodeHead(node->slaves, lnode->value);
  601. listDelNode(node_old->slaves, lnode);
  602. }
  603. listRelease(node_old->slaves);
  604. node_old->slaves = NULL;
  605. }
  606. listAddNodeHead(node->slaves, node_old);
  607. dictSetHashVal(*nodes, di, node);
  608. }
  609. else if(node->role == REDIS_ROLE_SLAVE)
  610. {
  611. if(node_old->slaves == NULL)
  612. {
  613. node_old->slaves = listCreate();
  614. if(node_old->slaves == NULL)
  615. {
  616. __redisClusterSetError(cc,REDIS_ERR_OOM,
  617. "Out of memory");
  618. return REDIS_ERR;
  619. }
  620. node_old->slaves->free =
  621. listClusterNodeDestructor;
  622. }
  623. listAddNodeTail(node_old->slaves, node);
  624. }
  625. else
  626. {
  627. NOT_REACHED();
  628. }
  629. }
  630. return REDIS_OK;
  631. }
  632. /**
  633. * Parse the "cluster slots" command reply to nodes dict.
  634. */
  635. dict *
  636. parse_cluster_slots(redisClusterContext *cc,
  637. redisReply *reply, int flags)
  638. {
  639. int ret;
  640. cluster_slot *slot = NULL;
  641. dict *nodes = NULL;
  642. dictEntry *den;
  643. redisReply *elem_slots;
  644. redisReply *elem_slots_begin, *elem_slots_end;
  645. redisReply *elem_nodes;
  646. redisReply *elem_ip, *elem_port;
  647. cluster_node *master = NULL, *slave;
  648. sds address;
  649. uint32_t i, idx;
  650. if(reply == NULL){
  651. return NULL;
  652. }
  653. nodes = dictCreate(&clusterNodesDictType, NULL);
  654. if(nodes == NULL){
  655. __redisClusterSetError(cc,REDIS_ERR_OOM,
  656. "out of memory");
  657. goto error;
  658. }
  659. if(reply->type != REDIS_REPLY_ARRAY || reply->elements <= 0){
  660. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  661. "Command(cluster slots) reply error: "
  662. "reply is not an array.");
  663. goto error;
  664. }
  665. for(i = 0; i < reply->elements; i ++){
  666. elem_slots = reply->element[i];
  667. if(elem_slots->type != REDIS_REPLY_ARRAY ||
  668. elem_slots->elements < 3){
  669. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  670. "Command(cluster slots) reply error: "
  671. "first sub_reply is not an array.");
  672. goto error;
  673. }
  674. slot = cluster_slot_create(NULL);
  675. if(slot == NULL){
  676. __redisClusterSetError(cc, REDIS_ERR_OOM,
  677. "Slot create failed: out of memory.");
  678. goto error;
  679. }
  680. //one slots region
  681. for(idx = 0; idx < elem_slots->elements; idx ++){
  682. if(idx == 0){
  683. elem_slots_begin = elem_slots->element[idx];
  684. if(elem_slots_begin->type != REDIS_REPLY_INTEGER){
  685. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  686. "Command(cluster slots) reply error: "
  687. "slot begin is not an integer.");
  688. goto error;
  689. }
  690. slot->start = (int)(elem_slots_begin->integer);
  691. }else if(idx == 1){
  692. elem_slots_end = elem_slots->element[idx];
  693. if(elem_slots_end->type != REDIS_REPLY_INTEGER){
  694. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  695. "Command(cluster slots) reply error: "
  696. "slot end is not an integer.");
  697. goto error;
  698. }
  699. slot->end = (int)(elem_slots_end->integer);
  700. if(slot->start > slot->end){
  701. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  702. "Command(cluster slots) reply error: "
  703. "slot begin is bigger than slot end.");
  704. goto error;
  705. }
  706. }else{
  707. elem_nodes = elem_slots->element[idx];
  708. if(elem_nodes->type != REDIS_REPLY_ARRAY ||
  709. elem_nodes->elements != 3){
  710. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  711. "Command(cluster slots) reply error: "
  712. "nodes sub_reply is not an correct array.");
  713. goto error;
  714. }
  715. elem_ip = elem_nodes->element[0];
  716. elem_port = elem_nodes->element[1];
  717. if(elem_ip == NULL || elem_port == NULL ||
  718. elem_ip->type != REDIS_REPLY_STRING ||
  719. elem_port->type != REDIS_REPLY_INTEGER){
  720. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  721. "Command(cluster slots) reply error: "
  722. "master ip or port is not correct.");
  723. goto error;
  724. }
  725. //this is master.
  726. if(idx == 2){
  727. address = sdsnewlen(elem_ip->str, elem_ip->len);
  728. address = sdscatfmt(address, ":%i", elem_port->integer);
  729. den = dictFind(nodes, address);
  730. //master already exits, break to the next slots region.
  731. if(den != NULL){
  732. sdsfree(address);
  733. master = dictGetEntryVal(den);
  734. ret = cluster_slot_ref_node(slot, master);
  735. if(ret != REDIS_OK){
  736. __redisClusterSetError(cc, REDIS_ERR_OOM,
  737. "Slot ref node failed: out of memory.");
  738. goto error;
  739. }
  740. slot = NULL;
  741. break;
  742. }
  743. sdsfree(address);
  744. master = node_get_with_slots(cc, elem_ip,
  745. elem_port, REDIS_ROLE_MASTER);
  746. if(master == NULL){
  747. goto error;
  748. }
  749. ret = dictAdd(nodes,
  750. sdsnewlen(master->addr, sdslen(master->addr)), master);
  751. if(ret != DICT_OK){
  752. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  753. "The address already exists in the nodes");
  754. cluster_node_deinit(master);
  755. hi_free(master);
  756. goto error;
  757. }
  758. ret = cluster_slot_ref_node(slot, master);
  759. if(ret != REDIS_OK){
  760. __redisClusterSetError(cc, REDIS_ERR_OOM,
  761. "Slot ref node failed: out of memory.");
  762. goto error;
  763. }
  764. slot = NULL;
  765. }else if(flags & HIRCLUSTER_FLAG_ADD_SLAVE){
  766. slave = node_get_with_slots(cc, elem_ip,
  767. elem_port, REDIS_ROLE_SLAVE);
  768. if(slave == NULL){
  769. goto error;
  770. }
  771. if(master->slaves == NULL){
  772. master->slaves = listCreate();
  773. if(master->slaves == NULL){
  774. __redisClusterSetError(cc,REDIS_ERR_OOM,
  775. "Out of memory");
  776. cluster_node_deinit(slave);
  777. goto error;
  778. }
  779. master->slaves->free =
  780. listClusterNodeDestructor;
  781. }
  782. listAddNodeTail(master->slaves, slave);
  783. }
  784. }
  785. }
  786. }
  787. return nodes;
  788. error:
  789. if(nodes != NULL){
  790. dictRelease(nodes);
  791. }
  792. if(slot != NULL){
  793. cluster_slot_destroy(slot);
  794. }
  795. return NULL;
  796. }
  797. /**
  798. * Parse the "cluster nodes" command reply to nodes dict.
  799. */
  800. dict *
  801. parse_cluster_nodes(redisClusterContext *cc,
  802. char *str, int str_len, int flags)
  803. {
  804. int ret;
  805. dict *nodes = NULL;
  806. dict *nodes_name = NULL;
  807. cluster_node *master, *slave;
  808. cluster_slot *slot;
  809. char *pos, *start, *end, *line_start, *line_end;
  810. char *role;
  811. int role_len;
  812. uint8_t myself = 0;
  813. int slot_start, slot_end;
  814. sds *part = NULL, *slot_start_end = NULL;
  815. int count_part = 0, count_slot_start_end = 0;
  816. int k;
  817. int len;
  818. nodes = dictCreate(&clusterNodesDictType, NULL);
  819. if(nodes == NULL){
  820. __redisClusterSetError(cc,REDIS_ERR_OOM,
  821. "out of memory");
  822. goto error;
  823. }
  824. start = str;
  825. end = start + str_len;
  826. line_start = start;
  827. for(pos = start; pos < end; pos ++){
  828. if(*pos == '\n'){
  829. line_end = pos - 1;
  830. len = line_end - line_start;
  831. part = sdssplitlen(line_start, len + 1, " ", 1, &count_part);
  832. if(part == NULL || count_part < 8){
  833. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  834. "split cluster nodes error");
  835. goto error;
  836. }
  837. //the address string is ":0", skip this node.
  838. if(sdslen(part[1]) == 2 && strcmp(part[1], ":0") == 0){
  839. sdsfreesplitres(part, count_part);
  840. count_part = 0;
  841. part = NULL;
  842. start = pos + 1;
  843. line_start = start;
  844. pos = start;
  845. continue;
  846. }
  847. if(sdslen(part[2]) >= 7 && memcmp(part[2], "myself,", 7) == 0){
  848. role_len = sdslen(part[2]) - 7;
  849. role = part[2] + 7;
  850. myself = 1;
  851. }else{
  852. role_len = sdslen(part[2]);
  853. role = part[2];
  854. }
  855. //add master node
  856. if(role_len >= 6 && memcmp(role, "master", 6) == 0){
  857. if(count_part < 8){
  858. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  859. "Master node parts number error: less than 8.");
  860. goto error;
  861. }
  862. master = node_get_with_nodes(cc,
  863. part, count_part, REDIS_ROLE_MASTER);
  864. if(master == NULL){
  865. goto error;
  866. }
  867. ret = dictAdd(nodes,
  868. sdsnewlen(master->addr, sdslen(master->addr)), master);
  869. if(ret != DICT_OK){
  870. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  871. "The address already exists in the nodes");
  872. cluster_node_deinit(master);
  873. hi_free(master);
  874. goto error;
  875. }
  876. if(flags & HIRCLUSTER_FLAG_ADD_SLAVE){
  877. ret = cluster_master_slave_mapping_with_name(cc,
  878. &nodes_name, master, master->name);
  879. if(ret != REDIS_OK){
  880. cluster_node_deinit(master);
  881. hi_free(master);
  882. goto error;
  883. }
  884. }
  885. if(myself) master->myself = 1;
  886. for(k = 8; k < count_part; k ++){
  887. slot_start_end = sdssplitlen(part[k],
  888. sdslen(part[k]), "-", 1, &count_slot_start_end);
  889. if(slot_start_end == NULL){
  890. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  891. "split slot start end error(NULL)");
  892. goto error;
  893. }else if(count_slot_start_end == 1){
  894. slot_start =
  895. hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));
  896. slot_end = slot_start;
  897. }else if(count_slot_start_end == 2){
  898. slot_start =
  899. hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));;
  900. slot_end =
  901. hi_atoi(slot_start_end[1], sdslen(slot_start_end[1]));;
  902. }else{
  903. //add open slot for master
  904. if(flags & HIRCLUSTER_FLAG_ADD_OPENSLOT &&
  905. count_slot_start_end == 3 &&
  906. sdslen(slot_start_end[0]) > 1 &&
  907. sdslen(slot_start_end[1]) == 1 &&
  908. sdslen(slot_start_end[2]) > 1 &&
  909. slot_start_end[0][0] == '[' &&
  910. slot_start_end[2][sdslen(slot_start_end[2])-1] == ']'){
  911. copen_slot *oslot, **oslot_elem;
  912. sdsrange(slot_start_end[0], 1, -1);
  913. sdsrange(slot_start_end[2], 0, -2);
  914. if(slot_start_end[1][0] == '>'){
  915. oslot = cluster_open_slot_create(
  916. hi_atoi(slot_start_end[0],
  917. sdslen(slot_start_end[0])),
  918. 1, slot_start_end[2], master);
  919. if(oslot == NULL){
  920. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  921. "create open slot error");
  922. goto error;
  923. }
  924. if(master->migrating == NULL){
  925. master->migrating = hiarray_create(1, sizeof(oslot));
  926. if(master->migrating == NULL){
  927. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  928. "create migrating array error");
  929. cluster_open_slot_destroy(oslot);
  930. goto error;
  931. }
  932. }
  933. oslot_elem = hiarray_push(master->migrating);
  934. if(oslot_elem == NULL){
  935. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  936. "Push migrating array error: out of memory");
  937. cluster_open_slot_destroy(oslot);
  938. goto error;
  939. }
  940. *oslot_elem = oslot;
  941. }else if(slot_start_end[1][0] == '<'){
  942. oslot = cluster_open_slot_create(hi_atoi(slot_start_end[0],
  943. sdslen(slot_start_end[0])), 0, slot_start_end[2],
  944. master);
  945. if(oslot == NULL){
  946. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  947. "create open slot error");
  948. goto error;
  949. }
  950. if(master->importing == NULL){
  951. master->importing = hiarray_create(1, sizeof(oslot));
  952. if(master->importing == NULL){
  953. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  954. "create migrating array error");
  955. cluster_open_slot_destroy(oslot);
  956. goto error;
  957. }
  958. }
  959. oslot_elem = hiarray_push(master->importing);
  960. if(oslot_elem == NULL){
  961. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  962. "push migrating array error: out of memory");
  963. cluster_open_slot_destroy(oslot);
  964. goto error;
  965. }
  966. *oslot_elem = oslot;
  967. }
  968. }
  969. slot_start = -1;
  970. slot_end = -1;
  971. }
  972. sdsfreesplitres(slot_start_end, count_slot_start_end);
  973. count_slot_start_end = 0;
  974. slot_start_end = NULL;
  975. if(slot_start < 0 || slot_end < 0 ||
  976. slot_start > slot_end || slot_end >= REDIS_CLUSTER_SLOTS){
  977. continue;
  978. }
  979. slot = cluster_slot_create(master);
  980. if(slot == NULL){
  981. __redisClusterSetError(cc,REDIS_ERR_OOM,
  982. "Out of memory");
  983. goto error;
  984. }
  985. slot->start = (uint32_t)slot_start;
  986. slot->end = (uint32_t)slot_end;
  987. }
  988. }
  989. //add slave node
  990. else if((flags & HIRCLUSTER_FLAG_ADD_SLAVE) &&
  991. (role_len >= 5 && memcmp(role, "slave", 5) == 0)){
  992. slave = node_get_with_nodes(cc, part,
  993. count_part, REDIS_ROLE_SLAVE);
  994. if(slave == NULL){
  995. goto error;
  996. }
  997. ret = cluster_master_slave_mapping_with_name(cc,
  998. &nodes_name, slave, part[3]);
  999. if(ret != REDIS_OK){
  1000. cluster_node_deinit(slave);
  1001. hi_free(slave);
  1002. goto error;
  1003. }
  1004. if(myself) slave->myself = 1;
  1005. }
  1006. if(myself == 1){
  1007. myself = 0;
  1008. }
  1009. sdsfreesplitres(part, count_part);
  1010. count_part = 0;
  1011. part = NULL;
  1012. start = pos + 1;
  1013. line_start = start;
  1014. pos = start;
  1015. }
  1016. }
  1017. if(nodes_name != NULL){
  1018. dictRelease(nodes_name);
  1019. }
  1020. return nodes;
  1021. error:
  1022. if(part != NULL){
  1023. sdsfreesplitres(part, count_part);
  1024. count_part = 0;
  1025. part = NULL;
  1026. }
  1027. if(slot_start_end != NULL){
  1028. sdsfreesplitres(slot_start_end, count_slot_start_end);
  1029. count_slot_start_end = 0;
  1030. slot_start_end = NULL;
  1031. }
  1032. if(nodes != NULL){
  1033. dictRelease(nodes);
  1034. }
  1035. if(nodes_name != NULL){
  1036. dictRelease(nodes_name);
  1037. }
  1038. return NULL;
  1039. }
  1040. /**
  1041. * Update route with the "cluster nodes" or "cluster slots" command reply.
  1042. */
  1043. static int
  1044. cluster_update_route_by_addr(redisClusterContext *cc,
  1045. const char *ip, int port)
  1046. {
  1047. redisContext *c = NULL;
  1048. redisReply *reply = NULL;
  1049. dict *nodes = NULL;
  1050. struct hiarray *slots = NULL;
  1051. cluster_node *master;
  1052. cluster_slot *slot, **slot_elem;
  1053. dictIterator *dit = NULL;
  1054. dictEntry *den;
  1055. listIter *lit = NULL;
  1056. listNode *lnode;
  1057. cluster_node *table[REDIS_CLUSTER_SLOTS];
  1058. uint32_t j, k;
  1059. if(cc == NULL){
  1060. return REDIS_ERR;
  1061. }
  1062. if(ip == NULL || port <= 0){
  1063. __redisClusterSetError(cc,
  1064. REDIS_ERR_OTHER,"Ip or port error!");
  1065. goto error;
  1066. }
  1067. if(cc->timeout){
  1068. c = redisConnectWithTimeout(ip, port, *cc->timeout);
  1069. }else{
  1070. c = redisConnect(ip, port);
  1071. }
  1072. if (c == NULL){
  1073. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1074. "Init redis context error(return NULL)");
  1075. goto error;
  1076. }else if(c->err){
  1077. __redisClusterSetError(cc,c->err,c->errstr);
  1078. goto error;
  1079. }
  1080. if(cc->flags & HIRCLUSTER_FLAG_ROUTE_USE_SLOTS){
  1081. reply = redisCommand(c, REDIS_COMMAND_CLUSTER_SLOTS);
  1082. if(reply == NULL){
  1083. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1084. "Command(cluster slots) reply error(NULL).");
  1085. goto error;
  1086. }else if(reply->type != REDIS_REPLY_ARRAY){
  1087. if(reply->type == REDIS_REPLY_ERROR){
  1088. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1089. reply->str);
  1090. }else{
  1091. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1092. "Command(cluster slots) reply error: type is not array.");
  1093. }
  1094. goto error;
  1095. }
  1096. nodes = parse_cluster_slots(cc, reply, cc->flags);
  1097. }else{
  1098. reply = redisCommand(c, REDIS_COMMAND_CLUSTER_NODES);
  1099. if(reply == NULL){
  1100. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1101. "Command(cluster nodes) reply error(NULL).");
  1102. goto error;
  1103. }else if(reply->type != REDIS_REPLY_STRING){
  1104. if(reply->type == REDIS_REPLY_ERROR){
  1105. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1106. reply->str);
  1107. }else{
  1108. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1109. "Command(cluster nodes) reply error: type is not string.");
  1110. }
  1111. goto error;
  1112. }
  1113. nodes = parse_cluster_nodes(cc, reply->str, reply->len, cc->flags);
  1114. }
  1115. if(nodes == NULL){
  1116. goto error;
  1117. }
  1118. memset(table, 0, REDIS_CLUSTER_SLOTS*sizeof(cluster_node *));
  1119. slots = hiarray_create(dictSize(nodes), sizeof(cluster_slot*));
  1120. if(slots == NULL){
  1121. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1122. "Slots array create failed: out of memory");
  1123. goto error;
  1124. }
  1125. dit = dictGetIterator(nodes);
  1126. if(dit == NULL){
  1127. __redisClusterSetError(cc,REDIS_ERR_OOM,
  1128. "Dict get iterator failed: out of memory");
  1129. goto error;
  1130. }
  1131. while((den = dictNext(dit))){
  1132. master = dictGetEntryVal(den);
  1133. if(master->role != REDIS_ROLE_MASTER){
  1134. __redisClusterSetError(cc,REDIS_ERR_OOM,
  1135. "Node role must be master");
  1136. goto error;
  1137. }
  1138. if(master->slots == NULL){
  1139. continue;
  1140. }
  1141. lit = listGetIterator(master->slots, AL_START_HEAD);
  1142. if(lit == NULL){
  1143. __redisClusterSetError(cc, REDIS_ERR_OOM,
  1144. "List get iterator failed: out of memory");
  1145. goto error;
  1146. }
  1147. while((lnode = listNext(lit))){
  1148. slot = listNodeValue(lnode);
  1149. if(slot->start > slot->end ||
  1150. slot->end >= REDIS_CLUSTER_SLOTS){
  1151. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  1152. "Slot region for node is error");
  1153. goto error;
  1154. }
  1155. slot_elem = hiarray_push(slots);
  1156. *slot_elem = slot;
  1157. }
  1158. listReleaseIterator(lit);
  1159. }
  1160. dictReleaseIterator(dit);
  1161. hiarray_sort(slots, cluster_slot_start_cmp);
  1162. for(j = 0; j < hiarray_n(slots); j ++){
  1163. slot_elem = hiarray_get(slots, j);
  1164. for(k = (*slot_elem)->start; k <= (*slot_elem)->end; k ++){
  1165. if(table[k] != NULL){
  1166. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  1167. "Diffent node hold a same slot");
  1168. goto error;
  1169. }
  1170. table[k] = (*slot_elem)->node;
  1171. }
  1172. }
  1173. cluster_nodes_swap_ctx(cc->nodes, nodes);
  1174. if(cc->nodes != NULL){
  1175. dictRelease(cc->nodes);
  1176. cc->nodes = NULL;
  1177. }
  1178. cc->nodes = nodes;
  1179. if(cc->slots != NULL)
  1180. {
  1181. cc->slots->nelem = 0;
  1182. hiarray_destroy(cc->slots);
  1183. cc->slots = NULL;
  1184. }
  1185. cc->slots = slots;
  1186. memcpy(cc->table, table, REDIS_CLUSTER_SLOTS*sizeof(cluster_node *));
  1187. cc->route_version ++;
  1188. freeReplyObject(reply);
  1189. if(c != NULL){
  1190. redisFree(c);
  1191. }
  1192. return REDIS_OK;
  1193. error:
  1194. if(dit != NULL){
  1195. dictReleaseIterator(dit);
  1196. }
  1197. if(lit != NULL){
  1198. listReleaseIterator(lit);
  1199. }
  1200. if(slots != NULL)
  1201. {
  1202. if(slots == cc->slots)
  1203. {
  1204. cc->slots = NULL;
  1205. }
  1206. slots->nelem = 0;
  1207. hiarray_destroy(slots);
  1208. }
  1209. if(nodes != NULL){
  1210. if(nodes == cc->nodes){
  1211. cc->nodes = NULL;
  1212. }
  1213. dictRelease(nodes);
  1214. }
  1215. if(reply != NULL){
  1216. freeReplyObject(reply);
  1217. reply = NULL;
  1218. }
  1219. if(c != NULL){
  1220. redisFree(c);
  1221. }
  1222. return REDIS_ERR;
  1223. }
  1224. /**
  1225. * Update route with the "cluster nodes" command reply.
  1226. */
  1227. static int
  1228. cluster_update_route_with_nodes_old(redisClusterContext *cc,
  1229. const char *ip, int port)
  1230. {
  1231. int ret;
  1232. redisContext *c = NULL;
  1233. redisReply *reply = NULL;
  1234. struct hiarray *slots = NULL;
  1235. dict *nodes = NULL;
  1236. dict *nodes_name = NULL;
  1237. cluster_node *master, *slave;
  1238. cluster_slot **slot;
  1239. char *pos, *start, *end, *line_start, *line_end;
  1240. char *role;
  1241. int role_len;
  1242. uint8_t myself = 0;
  1243. int slot_start, slot_end;
  1244. sds *part = NULL, *slot_start_end = NULL;
  1245. int count_part = 0, count_slot_start_end = 0;
  1246. int j, k;
  1247. int len;
  1248. cluster_node *table[REDIS_CLUSTER_SLOTS] = {NULL};
  1249. if(cc == NULL)
  1250. {
  1251. return REDIS_ERR;
  1252. }
  1253. if(ip == NULL || port <= 0)
  1254. {
  1255. __redisClusterSetError(cc,
  1256. REDIS_ERR_OTHER,"ip or port error!");
  1257. goto error;
  1258. }
  1259. if(cc->timeout)
  1260. {
  1261. c = redisConnectWithTimeout(ip, port, *cc->timeout);
  1262. }
  1263. else
  1264. {
  1265. c = redisConnect(ip, port);
  1266. }
  1267. if (c == NULL)
  1268. {
  1269. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1270. "init redis context error(return NULL)");
  1271. goto error;
  1272. }
  1273. else if(c->err)
  1274. {
  1275. __redisClusterSetError(cc,c->err,c->errstr);
  1276. goto error;
  1277. }
  1278. reply = redisCommand(c, REDIS_COMMAND_CLUSTER_NODES);
  1279. if(reply == NULL)
  1280. {
  1281. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1282. "command(cluster nodes) reply error(NULL)");
  1283. goto error;
  1284. }
  1285. else if(reply->type != REDIS_REPLY_STRING)
  1286. {
  1287. if(reply->type == REDIS_REPLY_ERROR)
  1288. {
  1289. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1290. reply->str);
  1291. }
  1292. else
  1293. {
  1294. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1295. "command(cluster nodes) reply error(type is not string)");
  1296. }
  1297. goto error;
  1298. }
  1299. nodes = dictCreate(&clusterNodesDictType, NULL);
  1300. slots = hiarray_create(10, sizeof(cluster_slot*));
  1301. if(slots == NULL)
  1302. {
  1303. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1304. "array create error");
  1305. goto error;
  1306. }
  1307. start = reply->str;
  1308. end = start + reply->len;
  1309. line_start = start;
  1310. for(pos = start; pos < end; pos ++)
  1311. {
  1312. if(*pos == '\n')
  1313. {
  1314. line_end = pos - 1;
  1315. len = line_end - line_start;
  1316. part = sdssplitlen(line_start, len + 1, " ", 1, &count_part);
  1317. if(part == NULL || count_part < 8)
  1318. {
  1319. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1320. "split cluster nodes error");
  1321. goto error;
  1322. }
  1323. //the address string is ":0", skip this node.
  1324. if(sdslen(part[1]) == 2 && strcmp(part[1], ":0") == 0)
  1325. {
  1326. sdsfreesplitres(part, count_part);
  1327. count_part = 0;
  1328. part = NULL;
  1329. start = pos + 1;
  1330. line_start = start;
  1331. pos = start;
  1332. continue;
  1333. }
  1334. if(sdslen(part[2]) >= 7 && memcmp(part[2], "myself,", 7) == 0)
  1335. {
  1336. role_len = sdslen(part[2]) - 7;
  1337. role = part[2] + 7;
  1338. myself = 1;
  1339. }
  1340. else
  1341. {
  1342. role_len = sdslen(part[2]);
  1343. role = part[2];
  1344. }
  1345. //add master node
  1346. if(role_len >= 6 && memcmp(role, "master", 6) == 0)
  1347. {
  1348. if(count_part < 8)
  1349. {
  1350. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1351. "master node part number error");
  1352. goto error;
  1353. }
  1354. master = node_get_with_nodes(cc,
  1355. part, count_part, REDIS_ROLE_MASTER);
  1356. if(master == NULL)
  1357. {
  1358. goto error;
  1359. }
  1360. ret = dictAdd(nodes,
  1361. sdsnewlen(master->addr, sdslen(master->addr)), master);
  1362. if(ret != DICT_OK)
  1363. {
  1364. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1365. "the address already exists in the nodes");
  1366. cluster_node_deinit(master);
  1367. hi_free(master);
  1368. goto error;
  1369. }
  1370. if(cc->flags & HIRCLUSTER_FLAG_ADD_SLAVE)
  1371. {
  1372. ret = cluster_master_slave_mapping_with_name(cc,
  1373. &nodes_name, master, master->name);
  1374. if(ret != REDIS_OK)
  1375. {
  1376. cluster_node_deinit(master);
  1377. hi_free(master);
  1378. goto error;
  1379. }
  1380. }
  1381. if(myself == 1)
  1382. {
  1383. master->con = c;
  1384. c = NULL;
  1385. }
  1386. for(k = 8; k < count_part; k ++)
  1387. {
  1388. slot_start_end = sdssplitlen(part[k],
  1389. sdslen(part[k]), "-", 1, &count_slot_start_end);
  1390. if(slot_start_end == NULL)
  1391. {
  1392. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1393. "split slot start end error(NULL)");
  1394. goto error;
  1395. }
  1396. else if(count_slot_start_end == 1)
  1397. {
  1398. slot_start =
  1399. hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));
  1400. slot_end = slot_start;
  1401. }
  1402. else if(count_slot_start_end == 2)
  1403. {
  1404. slot_start =
  1405. hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));;
  1406. slot_end =
  1407. hi_atoi(slot_start_end[1], sdslen(slot_start_end[1]));;
  1408. }
  1409. else
  1410. {
  1411. slot_start = -1;
  1412. slot_end = -1;
  1413. }
  1414. sdsfreesplitres(slot_start_end, count_slot_start_end);
  1415. count_slot_start_end = 0;
  1416. slot_start_end = NULL;
  1417. if(slot_start < 0 || slot_end < 0 ||
  1418. slot_start > slot_end || slot_end >= REDIS_CLUSTER_SLOTS)
  1419. {
  1420. continue;
  1421. }
  1422. for(j = slot_start; j <= slot_end; j ++)
  1423. {
  1424. if(table[j] != NULL)
  1425. {
  1426. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1427. "diffent node hold a same slot");
  1428. goto error;
  1429. }
  1430. table[j] = master;
  1431. }
  1432. slot = hiarray_push(slots);
  1433. if(slot == NULL)
  1434. {
  1435. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  1436. "slot push in array error");
  1437. goto error;
  1438. }
  1439. *slot = cluster_slot_create(master);
  1440. if(*slot == NULL)
  1441. {
  1442. __redisClusterSetError(cc,REDIS_ERR_OOM,
  1443. "Out of memory");
  1444. goto error;
  1445. }
  1446. (*slot)->start = (uint32_t)slot_start;
  1447. (*slot)->end = (uint32_t)slot_end;
  1448. }
  1449. }
  1450. //add slave node
  1451. else if((cc->flags & HIRCLUSTER_FLAG_ADD_SLAVE) &&
  1452. (role_len >= 5 && memcmp(role, "slave", 5) == 0))
  1453. {
  1454. slave = node_get_with_nodes(cc, part,
  1455. count_part, REDIS_ROLE_SLAVE);
  1456. if(slave == NULL)
  1457. {
  1458. goto error;
  1459. }
  1460. ret = cluster_master_slave_mapping_with_name(cc,
  1461. &nodes_name, slave, part[3]);
  1462. if(ret != REDIS_OK)
  1463. {
  1464. cluster_node_deinit(slave);
  1465. hi_free(slave);
  1466. goto error;
  1467. }
  1468. if(myself == 1)
  1469. {
  1470. slave->con = c;
  1471. c = NULL;
  1472. }
  1473. }
  1474. if(myself == 1)
  1475. {
  1476. myself = 0;
  1477. }
  1478. sdsfreesplitres(part, count_part);
  1479. count_part = 0;
  1480. part = NULL;
  1481. start = pos + 1;
  1482. line_start = start;
  1483. pos = start;
  1484. }
  1485. }
  1486. if(cc->slots != NULL)
  1487. {
  1488. cc->slots->nelem = 0;
  1489. hiarray_destroy(cc->slots);
  1490. cc->slots = NULL;
  1491. }
  1492. cc->slots = slots;
  1493. cluster_nodes_swap_ctx(cc->nodes, nodes);
  1494. if(cc->nodes != NULL)
  1495. {
  1496. dictRelease(cc->nodes);
  1497. cc->nodes = NULL;
  1498. }
  1499. cc->nodes = nodes;
  1500. hiarray_sort(cc->slots, cluster_slot_start_cmp);
  1501. memcpy(cc->table, table, REDIS_CLUSTER_SLOTS*sizeof(cluster_node *));
  1502. cc->route_version ++;
  1503. freeReplyObject(reply);
  1504. if(c != NULL)
  1505. {
  1506. redisFree(c);
  1507. }
  1508. if(nodes_name != NULL)
  1509. {
  1510. dictRelease(nodes_name);
  1511. }
  1512. return REDIS_OK;
  1513. error:
  1514. if(part != NULL)
  1515. {
  1516. sdsfreesplitres(part, count_part);
  1517. count_part = 0;
  1518. part = NULL;
  1519. }
  1520. if(slot_start_end != NULL)
  1521. {
  1522. sdsfreesplitres(slot_start_end, count_slot_start_end);
  1523. count_slot_start_end = 0;
  1524. slot_start_end = NULL;
  1525. }
  1526. if(slots != NULL)
  1527. {
  1528. if(slots == cc->slots)
  1529. {
  1530. cc->slots = NULL;
  1531. }
  1532. slots->nelem = 0;
  1533. hiarray_destroy(slots);
  1534. }
  1535. if(nodes != NULL)
  1536. {
  1537. if(nodes == cc->nodes)
  1538. {
  1539. cc->nodes = NULL;
  1540. }
  1541. dictRelease(nodes);
  1542. }
  1543. if(nodes_name != NULL)
  1544. {
  1545. dictRelease(nodes_name);
  1546. }
  1547. if(reply != NULL)
  1548. {
  1549. freeReplyObject(reply);
  1550. reply = NULL;
  1551. }
  1552. if(c != NULL)
  1553. {
  1554. redisFree(c);
  1555. }
  1556. return REDIS_ERR;
  1557. }
  1558. int
  1559. cluster_update_route(redisClusterContext *cc)
  1560. {
  1561. int ret;
  1562. int flag_err_not_set = 1;
  1563. cluster_node *node;
  1564. dictIterator *it;
  1565. dictEntry *de;
  1566. if(cc == NULL)
  1567. {
  1568. return REDIS_ERR;
  1569. }
  1570. if(cc->ip != NULL && cc->port > 0)
  1571. {
  1572. ret = cluster_update_route_by_addr(cc, cc->ip, cc->port);
  1573. if(ret == REDIS_OK)
  1574. {
  1575. return REDIS_OK;
  1576. }
  1577. flag_err_not_set = 0;
  1578. }
  1579. if(cc->nodes == NULL)
  1580. {
  1581. if(flag_err_not_set)
  1582. {
  1583. __redisClusterSetError(cc, REDIS_ERR_OTHER, "no server address");
  1584. }
  1585. return REDIS_ERR;
  1586. }
  1587. it = dictGetIterator(cc->nodes);
  1588. while ((de = dictNext(it)) != NULL)
  1589. {
  1590. node = dictGetEntryVal(de);
  1591. if(node == NULL || node->host == NULL || node->port < 0)
  1592. {
  1593. continue;
  1594. }
  1595. ret = cluster_update_route_by_addr(cc, node->host, node->port);
  1596. if(ret == REDIS_OK)
  1597. {
  1598. if(cc->err)
  1599. {
  1600. cc->err = 0;
  1601. memset(cc->errstr, '\0', strlen(cc->errstr));
  1602. }
  1603. dictReleaseIterator(it);
  1604. return REDIS_OK;
  1605. }
  1606. flag_err_not_set = 0;
  1607. }
  1608. dictReleaseIterator(it);
  1609. if(flag_err_not_set)
  1610. {
  1611. __redisClusterSetError(cc, REDIS_ERR_OTHER, "no valid server address");
  1612. }
  1613. return REDIS_ERR;
  1614. }
  1615. static void print_cluster_node_list(redisClusterContext *cc)
  1616. {
  1617. dictIterator *di = NULL;
  1618. dictEntry *de;
  1619. listIter *it;
  1620. listNode *ln;
  1621. cluster_node *master, *slave;
  1622. hilist *slaves;
  1623. if(cc == NULL)
  1624. {
  1625. return;
  1626. }
  1627. di = dictGetIterator(cc->nodes);
  1628. printf("name\taddress\trole\tslaves\n");
  1629. while((de = dictNext(di)) != NULL) {
  1630. master = dictGetEntryVal(de);
  1631. printf("%s\t%s\t%d\t%s\n",master->name, master->addr,
  1632. master->role, master->slaves?"hava":"null");
  1633. slaves = master->slaves;
  1634. if(slaves == NULL)
  1635. {
  1636. continue;
  1637. }
  1638. it = listGetIterator(slaves, AL_START_HEAD);
  1639. while((ln = listNext(it)) != NULL)
  1640. {
  1641. slave = listNodeValue(ln);
  1642. printf("%s\t%s\t%d\t%s\n",slave->name, slave->addr,
  1643. slave->role, slave->slaves?"hava":"null");
  1644. }
  1645. listReleaseIterator(it);
  1646. printf("\n");
  1647. }
  1648. }
  1649. int test_cluster_update_route(redisClusterContext *cc)
  1650. {
  1651. int ret;
  1652. ret = cluster_update_route(cc);
  1653. //print_cluster_node_list(cc);
  1654. return ret;
  1655. }
  1656. static redisClusterContext *redisClusterContextInit(void) {
  1657. redisClusterContext *cc;
  1658. cc = calloc(1,sizeof(redisClusterContext));
  1659. if (cc == NULL)
  1660. return NULL;
  1661. cc->err = 0;
  1662. cc->errstr[0] = '\0';
  1663. cc->ip = NULL;
  1664. cc->port = 0;
  1665. cc->flags = 0;
  1666. cc->timeout = NULL;
  1667. cc->nodes = NULL;
  1668. cc->slots = NULL;
  1669. cc->max_redirect_count = CLUSTER_DEFAULT_MAX_REDIRECT_COUNT;
  1670. cc->retry_count = 0;
  1671. cc->requests = NULL;
  1672. cc->need_update_route = 0;
  1673. cc->update_route_time = 0LL;
  1674. cc->route_version = 0LL;
  1675. memset(cc->table, 0, REDIS_CLUSTER_SLOTS*sizeof(cluster_node *));
  1676. return cc;
  1677. }
  1678. void redisClusterFree(redisClusterContext *cc) {
  1679. if (cc == NULL)
  1680. return;
  1681. if(cc->ip)
  1682. {
  1683. sdsfree(cc->ip);
  1684. cc->ip = NULL;
  1685. }
  1686. if (cc->timeout)
  1687. {
  1688. free(cc->timeout);
  1689. }
  1690. memset(cc->table, 0, REDIS_CLUSTER_SLOTS*sizeof(cluster_node *));
  1691. if(cc->slots != NULL)
  1692. {
  1693. cc->slots->nelem = 0;
  1694. hiarray_destroy(cc->slots);
  1695. cc->slots = NULL;
  1696. }
  1697. if(cc->nodes != NULL)
  1698. {
  1699. dictRelease(cc->nodes);
  1700. }
  1701. if(cc->requests != NULL)
  1702. {
  1703. listRelease(cc->requests);
  1704. }
  1705. free(cc);
  1706. }
  1707. static int redisClusterAddNode(redisClusterContext *cc, const char *addr)
  1708. {
  1709. dictEntry *node_entry;
  1710. cluster_node *node;
  1711. sds *ip_port = NULL;
  1712. int ip_port_count = 0;
  1713. sds ip;
  1714. int port;
  1715. if(cc == NULL)
  1716. {
  1717. return REDIS_ERR;
  1718. }
  1719. if(cc->nodes == NULL)
  1720. {
  1721. cc->nodes = dictCreate(&clusterNodesDictType, NULL);
  1722. if(cc->nodes == NULL)
  1723. {
  1724. return REDIS_ERR;
  1725. }
  1726. }
  1727. node_entry = dictFind(cc->nodes, addr);
  1728. if(node_entry == NULL)
  1729. {
  1730. ip_port = sdssplitlen(addr, strlen(addr),
  1731. IP_PORT_SEPARATOR, strlen(IP_PORT_SEPARATOR), &ip_port_count);
  1732. if(ip_port == NULL || ip_port_count != 2 ||
  1733. sdslen(ip_port[0]) <= 0 || sdslen(ip_port[1]) <= 0)
  1734. {
  1735. if(ip_port != NULL)
  1736. {
  1737. sdsfreesplitres(ip_port, ip_port_count);
  1738. }
  1739. __redisClusterSetError(cc,REDIS_ERR_OTHER,"server address is error(correct is like: 127.0.0.1:1234)");
  1740. return REDIS_ERR;
  1741. }
  1742. ip = ip_port[0];
  1743. port = hi_atoi(ip_port[1], sdslen(ip_port[1]));
  1744. if(port <= 0)
  1745. {
  1746. sdsfreesplitres(ip_port, ip_port_count);
  1747. __redisClusterSetError(cc,REDIS_ERR_OTHER,"server port is error");
  1748. return REDIS_ERR;
  1749. }
  1750. sdsfree(ip_port[1]);
  1751. free(ip_port);
  1752. ip_port = NULL;
  1753. node = hi_alloc(sizeof(cluster_node));
  1754. if(node == NULL)
  1755. {
  1756. sdsfree(ip);
  1757. __redisClusterSetError(cc,REDIS_ERR_OTHER,"alloc cluster node error");
  1758. return REDIS_ERR;
  1759. }
  1760. cluster_node_init(node);
  1761. node->addr = sdsnew(addr);
  1762. if(node->addr == NULL)
  1763. {
  1764. sdsfree(ip);
  1765. hi_free(node);
  1766. __redisClusterSetError(cc,REDIS_ERR_OTHER,"new node address error");
  1767. return REDIS_ERR;
  1768. }
  1769. node->host = ip;
  1770. node->port = port;
  1771. dictAdd(cc->nodes, sdsnewlen(node->addr, sdslen(node->addr)), node);
  1772. }
  1773. return REDIS_OK;
  1774. }
  1775. /* Connect to a Redis cluster. On error the field error in the returned
  1776. * context will be set to the return value of the error function.
  1777. * When no set of reply functions is given, the default set will be used. */
  1778. static redisClusterContext *_redisClusterConnect(redisClusterContext *cc, const char *addrs) {
  1779. int ret;
  1780. sds *address = NULL;
  1781. int address_count = 0;
  1782. int i;
  1783. if(cc == NULL)
  1784. {
  1785. return NULL;
  1786. }
  1787. address = sdssplitlen(addrs, strlen(addrs), CLUSTER_ADDRESS_SEPARATOR,
  1788. strlen(CLUSTER_ADDRESS_SEPARATOR), &address_count);
  1789. if(address == NULL || address_count <= 0)
  1790. {
  1791. __redisClusterSetError(cc,REDIS_ERR_OTHER,"servers address is error(correct is like: 127.0.0.1:1234,127.0.0.2:5678)");
  1792. return cc;
  1793. }
  1794. for(i = 0; i < address_count; i ++)
  1795. {
  1796. ret = redisClusterAddNode(cc, address[i]);
  1797. if(ret != REDIS_OK)
  1798. {
  1799. sdsfreesplitres(address, address_count);
  1800. return cc;
  1801. }
  1802. }
  1803. sdsfreesplitres(address, address_count);
  1804. cluster_update_route(cc);
  1805. return cc;
  1806. }
  1807. redisClusterContext *redisClusterConnect(const char *addrs, int flags)
  1808. {
  1809. redisClusterContext *cc;
  1810. cc = redisClusterContextInit();
  1811. if(cc == NULL)
  1812. {
  1813. return NULL;
  1814. }
  1815. cc->flags |= REDIS_BLOCK;
  1816. if(flags)
  1817. {
  1818. cc->flags |= flags;
  1819. }
  1820. return _redisClusterConnect(cc, addrs);
  1821. }
  1822. redisClusterContext *redisClusterConnectWithTimeout(
  1823. const char *addrs, const struct timeval tv, int flags)
  1824. {
  1825. redisClusterContext *cc;
  1826. cc = redisClusterContextInit();
  1827. if(cc == NULL)
  1828. {
  1829. return NULL;
  1830. }
  1831. cc->flags |= REDIS_BLOCK;
  1832. if(flags)
  1833. {
  1834. cc->flags |= flags;
  1835. }
  1836. if (cc->timeout == NULL)
  1837. {
  1838. cc->timeout = malloc(sizeof(struct timeval));
  1839. }
  1840. memcpy(cc->timeout, &tv, sizeof(struct timeval));
  1841. return _redisClusterConnect(cc, addrs);
  1842. }
  1843. redisClusterContext *redisClusterConnectNonBlock(const char *addrs, int flags) {
  1844. redisClusterContext *cc;
  1845. cc = redisClusterContextInit();
  1846. if(cc == NULL)
  1847. {
  1848. return NULL;
  1849. }
  1850. cc->flags &= ~REDIS_BLOCK;
  1851. if(flags)
  1852. {
  1853. cc->flags |= flags;
  1854. }
  1855. return _redisClusterConnect(cc, addrs);
  1856. }
  1857. redisContext *ctx_get_by_node(cluster_node *node,
  1858. const struct timeval *timeout, int flags)
  1859. {
  1860. redisContext *c = NULL;
  1861. if(node == NULL)
  1862. {
  1863. return NULL;
  1864. }
  1865. c = node->con;
  1866. if(c != NULL)
  1867. {
  1868. if(c->err)
  1869. {
  1870. redisReconnect(c);
  1871. }
  1872. return c;
  1873. }
  1874. if(node->host == NULL || node->port <= 0)
  1875. {
  1876. return NULL;
  1877. }
  1878. if(flags & REDIS_BLOCK)
  1879. {
  1880. if(timeout)
  1881. {
  1882. c = redisConnectWithTimeout(node->host, node->port, *timeout);
  1883. }
  1884. else
  1885. {
  1886. c = redisConnect(node->host, node->port);
  1887. }
  1888. }
  1889. else
  1890. {
  1891. c = redisConnectNonBlock(node->host, node->port);
  1892. }
  1893. node->con = c;
  1894. return c;
  1895. }
  1896. static cluster_node *node_get_by_slot(redisClusterContext *cc, uint32_t slot_num)
  1897. {
  1898. struct hiarray *slots;
  1899. uint32_t slot_count;
  1900. cluster_slot **slot;
  1901. uint32_t middle, start, end;
  1902. uint8_t stop = 0;
  1903. if(cc == NULL)
  1904. {
  1905. return NULL;
  1906. }
  1907. if(slot_num >= REDIS_CLUSTER_SLOTS)
  1908. {
  1909. return NULL;
  1910. }
  1911. slots = cc->slots;
  1912. if(slots == NULL)
  1913. {
  1914. return NULL;
  1915. }
  1916. slot_count = hiarray_n(slots);
  1917. start = 0;
  1918. end = slot_count - 1;
  1919. middle = 0;
  1920. do{
  1921. if(start >= end)
  1922. {
  1923. stop = 1;
  1924. middle = end;
  1925. }
  1926. else
  1927. {
  1928. middle = start + (end - start)/2;
  1929. }
  1930. ASSERT(middle < slot_count);
  1931. slot = hiarray_get(slots, middle);
  1932. if((*slot)->start > slot_num)
  1933. {
  1934. end = middle - 1;
  1935. }
  1936. else if((*slot)->end < slot_num)
  1937. {
  1938. start = middle + 1;
  1939. }
  1940. else
  1941. {
  1942. return (*slot)->node;
  1943. }
  1944. }while(!stop);
  1945. printf("slot_num : %d\n", slot_num);
  1946. printf("slot_count : %d\n", slot_count);
  1947. printf("start : %d\n", start);
  1948. printf("end : %d\n", end);
  1949. printf("middle : %d\n", middle);
  1950. return NULL;
  1951. }
  1952. static cluster_node *node_get_by_table(redisClusterContext *cc, uint32_t slot_num)
  1953. {
  1954. if(cc == NULL)
  1955. {
  1956. return NULL;
  1957. }
  1958. if(slot_num >= REDIS_CLUSTER_SLOTS)
  1959. {
  1960. return NULL;
  1961. }
  1962. return cc->table[slot_num];
  1963. }
  1964. static cluster_node *node_get_witch_connected(redisClusterContext *cc)
  1965. {
  1966. dictIterator *di;
  1967. dictEntry *de;
  1968. struct cluster_node *node;
  1969. redisContext *c = NULL;
  1970. redisReply *reply = NULL;
  1971. if(cc == NULL || cc->nodes == NULL)
  1972. {
  1973. return NULL;
  1974. }
  1975. di = dictGetIterator(cc->nodes);
  1976. while((de = dictNext(di)) != NULL)
  1977. {
  1978. node = dictGetEntryVal(de);
  1979. if(node == NULL)
  1980. {
  1981. continue;
  1982. }
  1983. c = ctx_get_by_node(node, cc->timeout, REDIS_BLOCK);
  1984. if(c == NULL || c->err)
  1985. {
  1986. continue;
  1987. }
  1988. reply = redisCommand(c, REDIS_COMMAND_PING);
  1989. if(reply != NULL && reply->type == REDIS_REPLY_STATUS &&
  1990. reply->str != NULL && strcmp(reply->str, "PONG") == 0)
  1991. {
  1992. freeReplyObject(reply);
  1993. reply = NULL;
  1994. dictReleaseIterator(di);
  1995. return node;
  1996. }
  1997. else if(reply != NULL)
  1998. {
  1999. freeReplyObject(reply);
  2000. reply = NULL;
  2001. }
  2002. }
  2003. dictReleaseIterator(di);
  2004. return NULL;
  2005. }
  2006. static int slot_get_by_command(redisClusterContext *cc, char *cmd, int len)
  2007. {
  2008. struct cmd *command = NULL;
  2009. struct keypos *kp;
  2010. int key_count;
  2011. uint32_t i;
  2012. int slot_num = -1;
  2013. if(cc == NULL || cmd == NULL || len <= 0)
  2014. {
  2015. goto done;
  2016. }
  2017. command = command_get();
  2018. if(command == NULL)
  2019. {
  2020. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2021. goto done;
  2022. }
  2023. command->cmd = cmd;
  2024. command->clen = len;
  2025. redis_parse_cmd(command);
  2026. if(command->result != CMD_PARSE_OK)
  2027. {
  2028. __redisClusterSetError(cc, REDIS_ERR_PROTOCOL, "parse command error");
  2029. goto done;
  2030. }
  2031. key_count = hiarray_n(command->keys);
  2032. if(key_count <= 0)
  2033. {
  2034. __redisClusterSetError(cc, REDIS_ERR_OTHER, "no keys in command(must have keys for redis cluster mode)");
  2035. goto done;
  2036. }
  2037. else if(key_count == 1)
  2038. {
  2039. kp = hiarray_get(command->keys, 0);
  2040. slot_num = keyHashSlot(kp->start, kp->end - kp->start);
  2041. goto done;
  2042. }
  2043. for(i = 0; i < hiarray_n(command->keys); i ++)
  2044. {
  2045. kp = hiarray_get(command->keys, i);
  2046. slot_num = keyHashSlot(kp->start, kp->end - kp->start);
  2047. }
  2048. done:
  2049. if(command != NULL)
  2050. {
  2051. command->cmd = NULL;
  2052. command_destroy(command);
  2053. }
  2054. return slot_num;
  2055. }
  2056. /* Get the cluster config from one node.
  2057. * Return value: config_value string must free by usr.
  2058. */
  2059. static char * cluster_config_get(redisClusterContext *cc,
  2060. const char *config_name, int *config_value_len)
  2061. {
  2062. redisContext *c;
  2063. cluster_node *node;
  2064. redisReply *reply = NULL, *sub_reply;
  2065. char *config_value = NULL;
  2066. if(cc == NULL || config_name == NULL
  2067. || config_value_len == NULL)
  2068. {
  2069. return NULL;
  2070. }
  2071. node = node_get_witch_connected(cc);
  2072. if(node == NULL)
  2073. {
  2074. __redisClusterSetError(cc,
  2075. REDIS_ERR_OTHER, "no reachable node in cluster");
  2076. goto error;
  2077. }
  2078. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  2079. reply = redisCommand(c, "config get %s", config_name);
  2080. if(reply == NULL)
  2081. {
  2082. __redisClusterSetError(cc,
  2083. REDIS_ERR_OTHER, "reply for config get is null");
  2084. goto error;
  2085. }
  2086. if(reply->type != REDIS_REPLY_ARRAY)
  2087. {
  2088. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2089. "reply for config get type is not array");
  2090. goto error;
  2091. }
  2092. if(reply->elements != 2)
  2093. {
  2094. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2095. "reply for config get elements number is not 2");
  2096. goto error;
  2097. }
  2098. sub_reply = reply->element[0];
  2099. if(sub_reply == NULL || sub_reply->type != REDIS_REPLY_STRING)
  2100. {
  2101. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2102. "reply for config get config name is not string");
  2103. goto error;
  2104. }
  2105. if(strcmp(sub_reply->str, config_name))
  2106. {
  2107. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2108. "reply for config get config name is not we want");
  2109. goto error;
  2110. }
  2111. sub_reply = reply->element[1];
  2112. if(sub_reply == NULL || sub_reply->type != REDIS_REPLY_STRING)
  2113. {
  2114. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2115. "reply for config get config value type is not string");
  2116. goto error;
  2117. }
  2118. config_value = sub_reply->str;
  2119. *config_value_len = sub_reply->len;
  2120. sub_reply->str= NULL;
  2121. if(reply != NULL)
  2122. {
  2123. freeReplyObject(reply);
  2124. }
  2125. return config_value;
  2126. error:
  2127. if(reply != NULL)
  2128. {
  2129. freeReplyObject(reply);
  2130. }
  2131. return NULL;
  2132. }
  2133. /* Helper function for the redisClusterAppendCommand* family of functions.
  2134. *
  2135. * Write a formatted command to the output buffer. When this family
  2136. * is used, you need to call redisGetReply yourself to retrieve
  2137. * the reply (or replies in pub/sub).
  2138. */
  2139. static int __redisClusterAppendCommand(redisClusterContext *cc,
  2140. struct cmd *command) {
  2141. cluster_node *node;
  2142. redisContext *c = NULL;
  2143. if(cc == NULL || command == NULL)
  2144. {
  2145. return REDIS_ERR;
  2146. }
  2147. node = node_get_by_table(cc, (uint32_t)command->slot_num);
  2148. if(node == NULL)
  2149. {
  2150. __redisClusterSetError(cc, REDIS_ERR_OTHER, "node get by slot error");
  2151. return REDIS_ERR;
  2152. }
  2153. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  2154. if(c == NULL)
  2155. {
  2156. __redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node is null");
  2157. return REDIS_ERR;
  2158. }
  2159. else if(c->err)
  2160. {
  2161. __redisClusterSetError(cc, c->err, c->errstr);
  2162. return REDIS_ERR;
  2163. }
  2164. if (__redisAppendCommand(c, command->cmd, command->clen) != REDIS_OK)
  2165. {
  2166. __redisClusterSetError(cc, c->err, c->errstr);
  2167. return REDIS_ERR;
  2168. }
  2169. return REDIS_OK;
  2170. }
  2171. /* Helper function for the redisClusterGetReply* family of functions.
  2172. */
  2173. static int __redisClusterGetReply(redisClusterContext *cc, int slot_num, void **reply)
  2174. {
  2175. cluster_node *node;
  2176. redisContext *c;
  2177. if(cc == NULL || slot_num < 0 || reply == NULL)
  2178. {
  2179. return REDIS_ERR;
  2180. }
  2181. node = node_get_by_table(cc, (uint32_t)slot_num);
  2182. if(node == NULL)
  2183. {
  2184. __redisClusterSetError(cc, REDIS_ERR_OTHER, "node get by table is null");
  2185. return REDIS_ERR;
  2186. }
  2187. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  2188. if(c == NULL)
  2189. {
  2190. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2191. return REDIS_ERR;
  2192. }
  2193. else if(c->err)
  2194. {
  2195. if(cc->need_update_route == 0)
  2196. {
  2197. cc->retry_count ++;
  2198. if(cc->retry_count > cc->max_redirect_count)
  2199. {
  2200. cc->need_update_route = 1;
  2201. cc->retry_count = 0;
  2202. }
  2203. }
  2204. __redisClusterSetError(cc, c->err, c->errstr);
  2205. return REDIS_ERR;
  2206. }
  2207. if(redisGetReply(c, reply) != REDIS_OK)
  2208. {
  2209. __redisClusterSetError(cc, c->err, c->errstr);
  2210. return REDIS_ERR;
  2211. }
  2212. if(cluster_reply_error_type(*reply) == CLUSTER_ERR_MOVED)
  2213. {
  2214. cc->need_update_route = 1;
  2215. }
  2216. return REDIS_OK;
  2217. }
  2218. static cluster_node *node_get_by_ask_error_reply(
  2219. redisClusterContext *cc, redisReply *reply)
  2220. {
  2221. sds *part = NULL, *ip_port = NULL;
  2222. int part_len = 0, ip_port_len;
  2223. dictEntry *de;
  2224. cluster_node *node = NULL;
  2225. if(cc == NULL || reply == NULL)
  2226. {
  2227. return NULL;
  2228. }
  2229. if(cluster_reply_error_type(reply) != CLUSTER_ERR_ASK)
  2230. {
  2231. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2232. "reply is not ask error!");
  2233. return NULL;
  2234. }
  2235. part = sdssplitlen(reply->str, reply->len, " ", 1, &part_len);
  2236. if(part != NULL && part_len == 3)
  2237. {
  2238. ip_port = sdssplitlen(part[2], sdslen(part[2]),
  2239. ":", 1, &ip_port_len);
  2240. if(ip_port != NULL && ip_port_len == 2)
  2241. {
  2242. de = dictFind(cc->nodes, part[2]);
  2243. if(de == NULL)
  2244. {
  2245. node = hi_alloc(sizeof(cluster_node));
  2246. if(node == NULL)
  2247. {
  2248. __redisClusterSetError(cc,
  2249. REDIS_ERR_OOM, "Out of memory");
  2250. goto done;
  2251. }
  2252. cluster_node_init(node);
  2253. node->addr = part[1];
  2254. node->host = ip_port[0];
  2255. node->port = hi_atoi(ip_port[1], sdslen(ip_port[1]));
  2256. node->role = REDIS_ROLE_MASTER;
  2257. dictAdd(cc->nodes, sdsnewlen(node->addr, sdslen(node->addr)), node);
  2258. part = NULL;
  2259. ip_port = NULL;
  2260. }
  2261. else
  2262. {
  2263. node = de->val;
  2264. goto done;
  2265. }
  2266. }
  2267. else
  2268. {
  2269. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2270. "ask error reply address part parse error!");
  2271. goto done;
  2272. }
  2273. }
  2274. else
  2275. {
  2276. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2277. "ask error reply parse error!");
  2278. goto done;
  2279. }
  2280. done:
  2281. if(part != NULL)
  2282. {
  2283. sdsfreesplitres(part, part_len);
  2284. part = NULL;
  2285. }
  2286. if(ip_port != NULL)
  2287. {
  2288. sdsfreesplitres(ip_port, ip_port_len);
  2289. ip_port = NULL;
  2290. }
  2291. return node;
  2292. }
  2293. static void *redis_cluster_command_execute(redisClusterContext *cc,
  2294. struct cmd *command)
  2295. {
  2296. int ret;
  2297. void *reply = NULL;
  2298. cluster_node *node;
  2299. redisContext *c = NULL;
  2300. int error_type;
  2301. retry:
  2302. node = node_get_by_table(cc, (uint32_t)command->slot_num);
  2303. if(node == NULL)
  2304. {
  2305. __redisClusterSetError(cc, REDIS_ERR_OTHER, "node get by table error");
  2306. return NULL;
  2307. }
  2308. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  2309. if(c == NULL)
  2310. {
  2311. __redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node is null");
  2312. return NULL;
  2313. }
  2314. else if(c->err)
  2315. {
  2316. node = node_get_witch_connected(cc);
  2317. if(node == NULL)
  2318. {
  2319. __redisClusterSetError(cc, REDIS_ERR_OTHER, "no reachable node in cluster");
  2320. return NULL;
  2321. }
  2322. cc->retry_count ++;
  2323. if(cc->retry_count > cc->max_redirect_count)
  2324. {
  2325. __redisClusterSetError(cc, REDIS_ERR_CLUSTER_TOO_MANY_REDIRECT,
  2326. "too many cluster redirect");
  2327. return NULL;
  2328. }
  2329. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  2330. if(c == NULL)
  2331. {
  2332. __redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node error");
  2333. return NULL;
  2334. }
  2335. else if(c->err)
  2336. {
  2337. __redisClusterSetError(cc, c->err, c->errstr);
  2338. return NULL;
  2339. }
  2340. }
  2341. ask_retry:
  2342. if (__redisAppendCommand(c,command->cmd, command->clen) != REDIS_OK)
  2343. {
  2344. __redisClusterSetError(cc, c->err, c->errstr);
  2345. return NULL;
  2346. }
  2347. reply = __redisBlockForReply(c);
  2348. if(reply == NULL)
  2349. {
  2350. __redisClusterSetError(cc, c->err, c->errstr);
  2351. return NULL;
  2352. }
  2353. error_type = cluster_reply_error_type(reply);
  2354. if(error_type > CLUSTER_NOT_ERR && error_type < CLUSTER_ERR_SENTINEL)
  2355. {
  2356. cc->retry_count ++;
  2357. if(cc->retry_count > cc->max_redirect_count)
  2358. {
  2359. __redisClusterSetError(cc, REDIS_ERR_CLUSTER_TOO_MANY_REDIRECT,
  2360. "too many cluster redirect");
  2361. freeReplyObject(reply);
  2362. return NULL;
  2363. }
  2364. switch(error_type)
  2365. {
  2366. case CLUSTER_ERR_MOVED:
  2367. freeReplyObject(reply);
  2368. reply = NULL;
  2369. ret = cluster_update_route(cc);
  2370. if(ret != REDIS_OK)
  2371. {
  2372. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  2373. "route update error, please recreate redisClusterContext!");
  2374. return NULL;
  2375. }
  2376. goto retry;
  2377. break;
  2378. case CLUSTER_ERR_ASK:
  2379. node = node_get_by_ask_error_reply(cc, reply);
  2380. if(node == NULL)
  2381. {
  2382. freeReplyObject(reply);
  2383. return NULL;
  2384. }
  2385. freeReplyObject(reply);
  2386. reply = NULL;
  2387. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  2388. if(c == NULL)
  2389. {
  2390. __redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node error");
  2391. return NULL;
  2392. }
  2393. else if(c->err)
  2394. {
  2395. __redisClusterSetError(cc, c->err, c->errstr);
  2396. return NULL;
  2397. }
  2398. reply = redisCommand(c, REDIS_COMMAND_ASKING);
  2399. if(reply == NULL)
  2400. {
  2401. __redisClusterSetError(cc, c->err, c->errstr);
  2402. return NULL;
  2403. }
  2404. freeReplyObject(reply);
  2405. reply = NULL;
  2406. goto ask_retry;
  2407. break;
  2408. case CLUSTER_ERR_TRYAGAIN:
  2409. case CLUSTER_ERR_CROSSSLOT:
  2410. case CLUSTER_ERR_CLUSTERDOWN:
  2411. freeReplyObject(reply);
  2412. reply = NULL;
  2413. goto retry;
  2414. break;
  2415. default:
  2416. break;
  2417. }
  2418. }
  2419. return reply;
  2420. }
  2421. static int command_pre_fragment(redisClusterContext *cc,
  2422. struct cmd *command, hilist *commands)
  2423. {
  2424. struct keypos *kp, *sub_kp;
  2425. uint32_t key_count;
  2426. uint32_t i, j;
  2427. uint32_t idx;
  2428. uint32_t key_len;
  2429. int slot_num = -1;
  2430. struct cmd *sub_command;
  2431. struct cmd **sub_commands = NULL;
  2432. char num_str[12];
  2433. uint8_t num_str_len;
  2434. if(command == NULL || commands == NULL)
  2435. {
  2436. goto done;
  2437. }
  2438. key_count = hiarray_n(command->keys);
  2439. sub_commands = hi_zalloc(REDIS_CLUSTER_SLOTS * sizeof(*sub_commands));
  2440. if (sub_commands == NULL)
  2441. {
  2442. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2443. goto done;
  2444. }
  2445. command->frag_seq = hi_alloc(key_count * sizeof(*command->frag_seq));
  2446. if(command->frag_seq == NULL)
  2447. {
  2448. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2449. goto done;
  2450. }
  2451. for(i = 0; i < key_count; i ++)
  2452. {
  2453. kp = hiarray_get(command->keys, i);
  2454. slot_num = keyHashSlot(kp->start, kp->end - kp->start);
  2455. if(slot_num < 0 || slot_num >= REDIS_CLUSTER_SLOTS)
  2456. {
  2457. __redisClusterSetError(cc,REDIS_ERR_OTHER,"keyHashSlot return error");
  2458. goto done;
  2459. }
  2460. if (sub_commands[slot_num] == NULL) {
  2461. sub_commands[slot_num] = command_get();
  2462. if (sub_commands[slot_num] == NULL) {
  2463. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2464. slot_num = -1;
  2465. goto done;
  2466. }
  2467. }
  2468. command->frag_seq[i] = sub_command = sub_commands[slot_num];
  2469. sub_command->narg++;
  2470. sub_kp = hiarray_push(sub_command->keys);
  2471. if (sub_kp == NULL) {
  2472. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2473. slot_num = -1;
  2474. goto done;
  2475. }
  2476. sub_kp->start = kp->start;
  2477. sub_kp->end = kp->end;
  2478. key_len = (uint32_t)(kp->end - kp->start);
  2479. sub_command->clen += key_len + uint_len(key_len);
  2480. sub_command->slot_num = slot_num;
  2481. if (command->type == CMD_REQ_REDIS_MSET) {
  2482. uint32_t len = 0;
  2483. char *p;
  2484. for (p = sub_kp->end + 1; !isdigit(*p); p++){}
  2485. p = sub_kp->end + 1;
  2486. while(!isdigit(*p))
  2487. {
  2488. p ++;
  2489. }
  2490. for (; isdigit(*p); p++) {
  2491. len = len * 10 + (uint32_t)(*p - '0');
  2492. }
  2493. len += CRLF_LEN * 2;
  2494. len += (p - sub_kp->end);
  2495. sub_kp->remain_len = len;
  2496. sub_command->clen += len;
  2497. }
  2498. }
  2499. for (i = 0; i < REDIS_CLUSTER_SLOTS; i++) { /* prepend command header */
  2500. sub_command = sub_commands[i];
  2501. if (sub_command == NULL) {
  2502. continue;
  2503. }
  2504. idx = 0;
  2505. if (command->type == CMD_REQ_REDIS_MGET) {
  2506. //"*%d\r\n$4\r\nmget\r\n"
  2507. sub_command->clen += 5*sub_command->narg;
  2508. sub_command->narg ++;
  2509. hi_itoa(num_str, sub_command->narg);
  2510. num_str_len = (uint8_t)(strlen(num_str));
  2511. sub_command->clen += 13 + num_str_len;
  2512. sub_command->cmd = hi_zalloc(sub_command->clen * sizeof(*sub_command->cmd));
  2513. if(sub_command->cmd == NULL)
  2514. {
  2515. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2516. slot_num = -1;
  2517. goto done;
  2518. }
  2519. sub_command->cmd[idx++] = '*';
  2520. memcpy(sub_command->cmd + idx, num_str, num_str_len);
  2521. idx += num_str_len;
  2522. memcpy(sub_command->cmd + idx, "\r\n$4\r\nmget\r\n", 12);
  2523. idx += 12;
  2524. for(j = 0; j < hiarray_n(sub_command->keys); j ++)
  2525. {
  2526. kp = hiarray_get(sub_command->keys, j);
  2527. key_len = (uint32_t)(kp->end - kp->start);
  2528. hi_itoa(num_str, key_len);
  2529. num_str_len = strlen(num_str);
  2530. sub_command->cmd[idx++] = '$';
  2531. memcpy(sub_command->cmd + idx, num_str, num_str_len);
  2532. idx += num_str_len;
  2533. memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
  2534. idx += CRLF_LEN;
  2535. memcpy(sub_command->cmd + idx, kp->start, key_len);
  2536. idx += key_len;
  2537. memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
  2538. idx += CRLF_LEN;
  2539. }
  2540. } else if (command->type == CMD_REQ_REDIS_DEL) {
  2541. //"*%d\r\n$3\r\ndel\r\n"
  2542. sub_command->clen += 5*sub_command->narg;
  2543. sub_command->narg ++;
  2544. hi_itoa(num_str, sub_command->narg);
  2545. num_str_len = (uint8_t)strlen(num_str);
  2546. sub_command->clen += 12 + num_str_len;
  2547. sub_command->cmd = hi_zalloc(sub_command->clen * sizeof(*sub_command->cmd));
  2548. if(sub_command->cmd == NULL)
  2549. {
  2550. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2551. slot_num = -1;
  2552. goto done;
  2553. }
  2554. sub_command->cmd[idx++] = '*';
  2555. memcpy(sub_command->cmd + idx, num_str, num_str_len);
  2556. idx += num_str_len;
  2557. memcpy(sub_command->cmd + idx, "\r\n$3\r\ndel\r\n", 11);
  2558. idx += 11;
  2559. for(j = 0; j < hiarray_n(sub_command->keys); j ++)
  2560. {
  2561. kp = hiarray_get(sub_command->keys, j);
  2562. key_len = (uint32_t)(kp->end - kp->start);
  2563. hi_itoa(num_str, key_len);
  2564. num_str_len = strlen(num_str);
  2565. sub_command->cmd[idx++] = '$';
  2566. memcpy(sub_command->cmd + idx, num_str, num_str_len);
  2567. idx += num_str_len;
  2568. memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
  2569. idx += CRLF_LEN;
  2570. memcpy(sub_command->cmd + idx, kp->start, key_len);
  2571. idx += key_len;
  2572. memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
  2573. idx += CRLF_LEN;
  2574. }
  2575. } else if (command->type == CMD_REQ_REDIS_MSET) {
  2576. //"*%d\r\n$4\r\nmset\r\n"
  2577. sub_command->clen += 3*sub_command->narg;
  2578. sub_command->narg *= 2;
  2579. sub_command->narg ++;
  2580. hi_itoa(num_str, sub_command->narg);
  2581. num_str_len = (uint8_t)strlen(num_str);
  2582. sub_command->clen += 13 + num_str_len;
  2583. sub_command->cmd = hi_zalloc(sub_command->clen * sizeof(*sub_command->cmd));
  2584. if(sub_command->cmd == NULL)
  2585. {
  2586. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2587. slot_num = -1;
  2588. goto done;
  2589. }
  2590. sub_command->cmd[idx++] = '*';
  2591. memcpy(sub_command->cmd + idx, num_str, num_str_len);
  2592. idx += num_str_len;
  2593. memcpy(sub_command->cmd + idx, "\r\n$4\r\nmset\r\n", 12);
  2594. idx += 12;
  2595. for(j = 0; j < hiarray_n(sub_command->keys); j ++)
  2596. {
  2597. kp = hiarray_get(sub_command->keys, j);
  2598. key_len = (uint32_t)(kp->end - kp->start);
  2599. hi_itoa(num_str, key_len);
  2600. num_str_len = strlen(num_str);
  2601. sub_command->cmd[idx++] = '$';
  2602. memcpy(sub_command->cmd + idx, num_str, num_str_len);
  2603. idx += num_str_len;
  2604. memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
  2605. idx += CRLF_LEN;
  2606. memcpy(sub_command->cmd + idx, kp->start, key_len + kp->remain_len);
  2607. idx += key_len + kp->remain_len;
  2608. }
  2609. } else {
  2610. NOT_REACHED();
  2611. }
  2612. //printf("len : %d\n", sub_command->clen);
  2613. //print_string_with_length_fix_CRLF(sub_command->cmd, sub_command->clen);
  2614. sub_command->type = command->type;
  2615. listAddNodeTail(commands, sub_command);
  2616. }
  2617. done:
  2618. if(sub_commands != NULL)
  2619. {
  2620. hi_free(sub_commands);
  2621. }
  2622. if(slot_num >= 0 && commands != NULL
  2623. && listLength(commands) == 1)
  2624. {
  2625. listNode *list_node = listFirst(commands);
  2626. listDelNode(commands, list_node);
  2627. if(command->frag_seq)
  2628. {
  2629. hi_free(command->frag_seq);
  2630. command->frag_seq = NULL;
  2631. }
  2632. command->slot_num = slot_num;
  2633. }
  2634. return slot_num;
  2635. }
  2636. static void *command_post_fragment(redisClusterContext *cc,
  2637. struct cmd *command, hilist *commands)
  2638. {
  2639. struct cmd *sub_command;
  2640. listNode *list_node;
  2641. listIter *list_iter;
  2642. redisReply *reply, *sub_reply;
  2643. long long count = 0;
  2644. list_iter = listGetIterator(commands, AL_START_HEAD);
  2645. while((list_node = listNext(list_iter)) != NULL)
  2646. {
  2647. sub_command = list_node->value;
  2648. reply = sub_command->reply;
  2649. if(reply == NULL)
  2650. {
  2651. return NULL;
  2652. }
  2653. else if(reply->type == REDIS_REPLY_ERROR)
  2654. {
  2655. return reply;
  2656. }
  2657. if (command->type == CMD_REQ_REDIS_MGET) {
  2658. if(reply->type != REDIS_REPLY_ARRAY)
  2659. {
  2660. __redisClusterSetError(cc,REDIS_ERR_OTHER,"reply type is error(here only can be array)");
  2661. return NULL;
  2662. }
  2663. }else if(command->type == CMD_REQ_REDIS_DEL){
  2664. if(reply->type != REDIS_REPLY_INTEGER)
  2665. {
  2666. __redisClusterSetError(cc,REDIS_ERR_OTHER,"reply type is error(here only can be integer)");
  2667. return NULL;
  2668. }
  2669. count += reply->integer;
  2670. }else if(command->type == CMD_REQ_REDIS_MSET){
  2671. if(reply->type != REDIS_REPLY_STATUS ||
  2672. reply->len != 2 || strcmp(reply->str, REDIS_STATUS_OK) != 0)
  2673. {
  2674. __redisClusterSetError(cc,REDIS_ERR_OTHER,"reply type is error(here only can be status and ok)");
  2675. return NULL;
  2676. }
  2677. }else {
  2678. NOT_REACHED();
  2679. }
  2680. }
  2681. reply = hi_calloc(1,sizeof(*reply));
  2682. if (reply == NULL)
  2683. {
  2684. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2685. return NULL;
  2686. }
  2687. if (command->type == CMD_REQ_REDIS_MGET) {
  2688. int i;
  2689. uint32_t key_count;
  2690. reply->type = REDIS_REPLY_ARRAY;
  2691. key_count = hiarray_n(command->keys);
  2692. reply->elements = key_count;
  2693. reply->element = hi_calloc(key_count, sizeof(*reply));
  2694. if (reply->element == NULL) {
  2695. freeReplyObject(reply);
  2696. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2697. return NULL;
  2698. }
  2699. for (i = key_count - 1; i >= 0; i--) { /* for each key */
  2700. sub_reply = command->frag_seq[i]->reply; /* get it's reply */
  2701. if (sub_reply == NULL) {
  2702. freeReplyObject(reply);
  2703. __redisClusterSetError(cc,REDIS_ERR_OTHER,"sub reply is null");
  2704. return NULL;
  2705. }
  2706. if(sub_reply->type == REDIS_REPLY_STRING)
  2707. {
  2708. reply->element[i] = sub_reply;
  2709. }
  2710. else if(sub_reply->type == REDIS_REPLY_ARRAY)
  2711. {
  2712. if(sub_reply->elements == 0)
  2713. {
  2714. freeReplyObject(reply);
  2715. __redisClusterSetError(cc,REDIS_ERR_OTHER,"sub reply elements error");
  2716. return NULL;
  2717. }
  2718. reply->element[i] = sub_reply->element[sub_reply->elements - 1];
  2719. sub_reply->elements --;
  2720. }
  2721. }
  2722. }else if(command->type == CMD_REQ_REDIS_DEL){
  2723. reply->type = REDIS_REPLY_INTEGER;
  2724. reply->integer = count;
  2725. }else if(command->type == CMD_REQ_REDIS_MSET){
  2726. reply->type = REDIS_REPLY_STATUS;
  2727. uint32_t str_len = strlen(REDIS_STATUS_OK);
  2728. reply->str = hi_alloc((str_len + 1) * sizeof(char*));
  2729. if(reply->str == NULL)
  2730. {
  2731. freeReplyObject(reply);
  2732. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2733. return NULL;
  2734. }
  2735. reply->len = str_len;
  2736. memcpy(reply->str, REDIS_STATUS_OK, str_len);
  2737. reply->str[str_len] = '\0';
  2738. }else {
  2739. NOT_REACHED();
  2740. }
  2741. return reply;
  2742. }
  2743. /*
  2744. * Split the command into subcommands by slot
  2745. *
  2746. * Returns slot_num
  2747. * If slot_num < 0 or slot_num >= REDIS_CLUSTER_SLOTS means this function runs error;
  2748. * Otherwise if the commands > 1 , slot_num is the last subcommand slot number.
  2749. */
  2750. static int command_format_by_slot(redisClusterContext *cc,
  2751. struct cmd *command, hilist *commands)
  2752. {
  2753. struct keypos *kp;
  2754. int key_count;
  2755. int slot_num = -1;
  2756. if(cc == NULL || commands == NULL ||
  2757. command == NULL ||
  2758. command->cmd == NULL || command->clen <= 0)
  2759. {
  2760. goto done;
  2761. }
  2762. redis_parse_cmd(command);
  2763. if(command->result == CMD_PARSE_ENOMEM)
  2764. {
  2765. __redisClusterSetError(cc, REDIS_ERR_PROTOCOL, "Parse command error: out of memory");
  2766. goto done;
  2767. }
  2768. else if(command->result != CMD_PARSE_OK)
  2769. {
  2770. __redisClusterSetError(cc, REDIS_ERR_PROTOCOL, command->errstr);
  2771. goto done;
  2772. }
  2773. key_count = hiarray_n(command->keys);
  2774. if(key_count <= 0)
  2775. {
  2776. __redisClusterSetError(cc, REDIS_ERR_OTHER, "No keys in command(must have keys for redis cluster mode)");
  2777. goto done;
  2778. }
  2779. else if(key_count == 1)
  2780. {
  2781. kp = hiarray_get(command->keys, 0);
  2782. slot_num = keyHashSlot(kp->start, kp->end - kp->start);
  2783. command->slot_num = slot_num;
  2784. goto done;
  2785. }
  2786. slot_num = command_pre_fragment(cc, command, commands);
  2787. done:
  2788. return slot_num;
  2789. }
  2790. void redisClusterSetMaxRedirect(redisClusterContext *cc, int max_redirect_count)
  2791. {
  2792. if(cc == NULL || max_redirect_count <= 0)
  2793. {
  2794. return;
  2795. }
  2796. cc->max_redirect_count = max_redirect_count;
  2797. }
  2798. void *redisClusterFormattedCommand(redisClusterContext *cc, char *cmd, int len) {
  2799. redisReply *reply = NULL;
  2800. int slot_num;
  2801. struct cmd *command = NULL, *sub_command;
  2802. hilist *commands = NULL;
  2803. listNode *list_node;
  2804. listIter *list_iter = NULL;
  2805. if(cc == NULL)
  2806. {
  2807. return NULL;
  2808. }
  2809. if(cc->err)
  2810. {
  2811. cc->err = 0;
  2812. memset(cc->errstr, '\0', strlen(cc->errstr));
  2813. }
  2814. command = command_get();
  2815. if(command == NULL)
  2816. {
  2817. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2818. return NULL;
  2819. }
  2820. command->cmd = cmd;
  2821. command->clen = len;
  2822. commands = listCreate();
  2823. if(commands == NULL)
  2824. {
  2825. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2826. goto error;
  2827. }
  2828. commands->free = listCommandFree;
  2829. slot_num = command_format_by_slot(cc, command, commands);
  2830. if(slot_num < 0)
  2831. {
  2832. goto error;
  2833. }
  2834. else if(slot_num >= REDIS_CLUSTER_SLOTS)
  2835. {
  2836. __redisClusterSetError(cc,REDIS_ERR_OTHER,"slot_num is out of range");
  2837. goto error;
  2838. }
  2839. //all keys belong to one slot
  2840. if(listLength(commands) == 0)
  2841. {
  2842. reply = redis_cluster_command_execute(cc, command);
  2843. goto done;
  2844. }
  2845. ASSERT(listLength(commands) != 1);
  2846. list_iter = listGetIterator(commands, AL_START_HEAD);
  2847. while((list_node = listNext(list_iter)) != NULL)
  2848. {
  2849. sub_command = list_node->value;
  2850. reply = redis_cluster_command_execute(cc, sub_command);
  2851. if(reply == NULL)
  2852. {
  2853. goto error;
  2854. }
  2855. else if(reply->type == REDIS_REPLY_ERROR)
  2856. {
  2857. goto done;
  2858. }
  2859. sub_command->reply = reply;
  2860. }
  2861. reply = command_post_fragment(cc, command, commands);
  2862. done:
  2863. command->cmd = NULL;
  2864. command_destroy(command);
  2865. if(commands != NULL)
  2866. {
  2867. listRelease(commands);
  2868. }
  2869. if(list_iter != NULL)
  2870. {
  2871. listReleaseIterator(list_iter);
  2872. }
  2873. cc->retry_count = 0;
  2874. return reply;
  2875. error:
  2876. if(command != NULL)
  2877. {
  2878. command->cmd = NULL;
  2879. command_destroy(command);
  2880. }
  2881. if(commands != NULL)
  2882. {
  2883. listRelease(commands);
  2884. }
  2885. if(list_iter != NULL)
  2886. {
  2887. listReleaseIterator(list_iter);
  2888. }
  2889. cc->retry_count = 0;
  2890. return NULL;
  2891. }
  2892. void *redisClustervCommand(redisClusterContext *cc, const char *format, va_list ap) {
  2893. redisReply *reply;
  2894. char *cmd;
  2895. int len;
  2896. if(cc == NULL)
  2897. {
  2898. return NULL;
  2899. }
  2900. len = redisvFormatCommand(&cmd,format,ap);
  2901. if (len == -1) {
  2902. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2903. return NULL;
  2904. } else if (len == -2) {
  2905. __redisClusterSetError(cc,REDIS_ERR_OTHER,"Invalid format string");
  2906. return NULL;
  2907. }
  2908. reply = redisClusterFormattedCommand(cc, cmd, len);
  2909. free(cmd);
  2910. return reply;
  2911. }
  2912. void *redisClusterCommand(redisClusterContext *cc, const char *format, ...) {
  2913. va_list ap;
  2914. redisReply *reply = NULL;
  2915. va_start(ap,format);
  2916. reply = redisClustervCommand(cc, format, ap);
  2917. va_end(ap);
  2918. return reply;
  2919. }
  2920. void *redisClusterCommandArgv(redisClusterContext *cc, int argc, const char **argv, const size_t *argvlen) {
  2921. redisReply *reply = NULL;
  2922. char *cmd;
  2923. int len;
  2924. len = redisFormatCommandArgv(&cmd,argc,argv,argvlen);
  2925. if (len == -1) {
  2926. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2927. return NULL;
  2928. }
  2929. reply = redisClusterFormattedCommand(cc, cmd, len);
  2930. free(cmd);
  2931. return reply;
  2932. }
  2933. int redisClusterAppendFormattedCommand(redisClusterContext *cc,
  2934. char *cmd, int len) {
  2935. int slot_num;
  2936. struct cmd *command = NULL, *sub_command;
  2937. hilist *commands = NULL;
  2938. listNode *list_node;
  2939. listIter *list_iter = NULL;
  2940. if(cc->requests == NULL)
  2941. {
  2942. cc->requests = listCreate();
  2943. if(cc->requests == NULL)
  2944. {
  2945. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2946. goto error;
  2947. }
  2948. cc->requests->free = listCommandFree;
  2949. }
  2950. command = command_get();
  2951. if(command == NULL)
  2952. {
  2953. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2954. goto error;
  2955. }
  2956. command->cmd = cmd;
  2957. command->clen = len;
  2958. commands = listCreate();
  2959. if(commands == NULL)
  2960. {
  2961. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  2962. goto error;
  2963. }
  2964. commands->free = listCommandFree;
  2965. slot_num = command_format_by_slot(cc, command, commands);
  2966. if(slot_num < 0)
  2967. {
  2968. goto error;
  2969. }
  2970. else if(slot_num >= REDIS_CLUSTER_SLOTS)
  2971. {
  2972. __redisClusterSetError(cc,REDIS_ERR_OTHER,"slot_num is out of range");
  2973. goto error;
  2974. }
  2975. //all keys belong to one slot
  2976. if(listLength(commands) == 0)
  2977. {
  2978. if(__redisClusterAppendCommand(cc, command) == REDIS_OK)
  2979. {
  2980. goto done;
  2981. }
  2982. else
  2983. {
  2984. goto error;
  2985. }
  2986. }
  2987. ASSERT(listLength(commands) != 1);
  2988. list_iter = listGetIterator(commands, AL_START_HEAD);
  2989. while((list_node = listNext(list_iter)) != NULL)
  2990. {
  2991. sub_command = list_node->value;
  2992. if(__redisClusterAppendCommand(cc, sub_command) == REDIS_OK)
  2993. {
  2994. continue;
  2995. }
  2996. else
  2997. {
  2998. goto error;
  2999. }
  3000. }
  3001. done:
  3002. if(command->cmd != NULL)
  3003. {
  3004. command->cmd = NULL;
  3005. }
  3006. else
  3007. {
  3008. goto error;
  3009. }
  3010. if(commands != NULL)
  3011. {
  3012. if(listLength(commands) > 0)
  3013. {
  3014. command->sub_commands = commands;
  3015. }
  3016. else
  3017. {
  3018. listRelease(commands);
  3019. }
  3020. }
  3021. if(list_iter != NULL)
  3022. {
  3023. listReleaseIterator(list_iter);
  3024. }
  3025. listAddNodeTail(cc->requests, command);
  3026. return REDIS_OK;
  3027. error:
  3028. if(command != NULL)
  3029. {
  3030. command->cmd = NULL;
  3031. command_destroy(command);
  3032. }
  3033. if(commands != NULL)
  3034. {
  3035. listRelease(commands);
  3036. }
  3037. if(list_iter != NULL)
  3038. {
  3039. listReleaseIterator(list_iter);
  3040. }
  3041. /* Attention: mybe here we must pop the
  3042. sub_commands that had append to the nodes.
  3043. But now we do not handle it. */
  3044. return REDIS_ERR;
  3045. }
  3046. int redisClustervAppendCommand(redisClusterContext *cc,
  3047. const char *format, va_list ap) {
  3048. int ret;
  3049. char *cmd;
  3050. int len;
  3051. len = redisvFormatCommand(&cmd,format,ap);
  3052. if (len == -1) {
  3053. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  3054. return REDIS_ERR;
  3055. } else if (len == -2) {
  3056. __redisClusterSetError(cc,REDIS_ERR_OTHER,"Invalid format string");
  3057. return REDIS_ERR;
  3058. }
  3059. ret = redisClusterAppendFormattedCommand(cc, cmd, len);
  3060. free(cmd);
  3061. return ret;
  3062. }
  3063. int redisClusterAppendCommand(redisClusterContext *cc,
  3064. const char *format, ...) {
  3065. int ret;
  3066. va_list ap;
  3067. if(cc == NULL || format == NULL)
  3068. {
  3069. return REDIS_ERR;
  3070. }
  3071. va_start(ap,format);
  3072. ret = redisClustervAppendCommand(cc, format, ap);
  3073. va_end(ap);
  3074. return ret;
  3075. }
  3076. int redisClusterAppendCommandArgv(redisClusterContext *cc,
  3077. int argc, const char **argv, const size_t *argvlen) {
  3078. int ret;
  3079. char *cmd;
  3080. int len;
  3081. len = redisFormatCommandArgv(&cmd,argc,argv,argvlen);
  3082. if (len == -1) {
  3083. __redisClusterSetError(cc,REDIS_ERR_OOM,"Out of memory");
  3084. return REDIS_ERR;
  3085. }
  3086. ret = redisClusterAppendFormattedCommand(cc, cmd, len);
  3087. free(cmd);
  3088. return ret;
  3089. }
  3090. static int redisCLusterSendAll(redisClusterContext *cc)
  3091. {
  3092. dictIterator *di;
  3093. dictEntry *de;
  3094. struct cluster_node *node;
  3095. redisContext *c = NULL;
  3096. int wdone = 0;
  3097. if(cc == NULL || cc->nodes == NULL)
  3098. {
  3099. return REDIS_ERR;
  3100. }
  3101. di = dictGetIterator(cc->nodes);
  3102. while((de = dictNext(di)) != NULL)
  3103. {
  3104. node = dictGetEntryVal(de);
  3105. if(node == NULL)
  3106. {
  3107. continue;
  3108. }
  3109. c = ctx_get_by_node(node, cc->timeout, cc->flags);
  3110. if(c == NULL)
  3111. {
  3112. continue;
  3113. }
  3114. if (c->flags & REDIS_BLOCK) {
  3115. /* Write until done */
  3116. do {
  3117. if (redisBufferWrite(c,&wdone) == REDIS_ERR)
  3118. {
  3119. dictReleaseIterator(di);
  3120. return REDIS_ERR;
  3121. }
  3122. } while (!wdone);
  3123. }
  3124. }
  3125. dictReleaseIterator(di);
  3126. return REDIS_OK;
  3127. }
  3128. static int redisCLusterClearAll(redisClusterContext *cc)
  3129. {
  3130. dictIterator *di;
  3131. dictEntry *de;
  3132. struct cluster_node *node;
  3133. redisContext *c = NULL;
  3134. if (cc == NULL) {
  3135. return REDIS_ERR;
  3136. }
  3137. if (cc->err) {
  3138. cc->err = 0;
  3139. memset(cc->errstr, '\0', strlen(cc->errstr));
  3140. }
  3141. if (cc->nodes == NULL) {
  3142. return REDIS_ERR;
  3143. }
  3144. di = dictGetIterator(cc->nodes);
  3145. while((de = dictNext(di)) != NULL)
  3146. {
  3147. node = dictGetEntryVal(de);
  3148. if(node == NULL)
  3149. {
  3150. continue;
  3151. }
  3152. c = node->con;
  3153. if(c == NULL)
  3154. {
  3155. continue;
  3156. }
  3157. redisFree(c);
  3158. node->con = NULL;
  3159. }
  3160. dictReleaseIterator(di);
  3161. return REDIS_OK;
  3162. }
  3163. int redisClusterGetReply(redisClusterContext *cc, void **reply) {
  3164. struct cmd *command, *sub_command;
  3165. hilist *commands = NULL;
  3166. listNode *list_command, *list_sub_command;
  3167. listIter *list_iter;
  3168. int slot_num;
  3169. void *sub_reply;
  3170. if(cc == NULL || reply == NULL)
  3171. return REDIS_ERR;
  3172. cc->err = 0;
  3173. cc->errstr[0] = '\0';
  3174. *reply = NULL;
  3175. if (cc->requests == NULL)
  3176. return REDIS_ERR;
  3177. list_command = listFirst(cc->requests);
  3178. //no more reply
  3179. if(list_command == NULL)
  3180. {
  3181. *reply = NULL;
  3182. return REDIS_OK;
  3183. }
  3184. command = list_command->value;
  3185. if(command == NULL)
  3186. {
  3187. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  3188. "command in the requests list is null");
  3189. goto error;
  3190. }
  3191. slot_num = command->slot_num;
  3192. if(slot_num >= 0)
  3193. {
  3194. listDelNode(cc->requests, list_command);
  3195. return __redisClusterGetReply(cc, slot_num, reply);
  3196. }
  3197. commands = command->sub_commands;
  3198. if(commands == NULL)
  3199. {
  3200. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  3201. "sub_commands in command is null");
  3202. goto error;
  3203. }
  3204. ASSERT(listLength(commands) != 1);
  3205. list_iter = listGetIterator(commands, AL_START_HEAD);
  3206. while((list_sub_command = listNext(list_iter)) != NULL)
  3207. {
  3208. sub_command = list_sub_command->value;
  3209. if(sub_command == NULL)
  3210. {
  3211. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  3212. "sub_command is null");
  3213. goto error;
  3214. }
  3215. slot_num = sub_command->slot_num;
  3216. if(slot_num < 0)
  3217. {
  3218. __redisClusterSetError(cc,REDIS_ERR_OTHER,
  3219. "sub_command slot_num is less then zero");
  3220. goto error;
  3221. }
  3222. if(__redisClusterGetReply(cc, slot_num, &sub_reply) != REDIS_OK)
  3223. {
  3224. goto error;
  3225. }
  3226. sub_command->reply = sub_reply;
  3227. }
  3228. *reply = command_post_fragment(cc, command, commands);
  3229. if(*reply == NULL)
  3230. {
  3231. goto error;
  3232. }
  3233. listDelNode(cc->requests, list_command);
  3234. return REDIS_OK;
  3235. error:
  3236. listDelNode(cc->requests, list_command);
  3237. return REDIS_ERR;
  3238. }
  3239. void redisClusterReset(redisClusterContext *cc)
  3240. {
  3241. int status;
  3242. void *reply;
  3243. if(cc == NULL || cc->nodes == NULL)
  3244. {
  3245. return;
  3246. }
  3247. if (cc->err) {
  3248. redisCLusterClearAll(cc);
  3249. } else {
  3250. redisCLusterSendAll(cc);
  3251. do {
  3252. status = redisClusterGetReply(cc, &reply);
  3253. if (status == REDIS_OK) {
  3254. freeReplyObject(reply);
  3255. } else {
  3256. redisCLusterClearAll(cc);
  3257. break;
  3258. }
  3259. } while(reply != NULL);
  3260. }
  3261. if(cc->requests)
  3262. {
  3263. listRelease(cc->requests);
  3264. cc->requests = NULL;
  3265. }
  3266. if(cc->need_update_route)
  3267. {
  3268. status = cluster_update_route(cc);
  3269. if(status != REDIS_OK)
  3270. {
  3271. __redisClusterSetError(cc, REDIS_ERR_OTHER,
  3272. "route update error, please recreate redisClusterContext!");
  3273. return;
  3274. }
  3275. cc->need_update_route = 0;
  3276. }
  3277. }
  3278. /*############redis cluster async############*/
  3279. /* We want the error field to be accessible directly instead of requiring
  3280. * an indirection to the redisContext struct. */
  3281. static void __redisClusterAsyncCopyError(redisClusterAsyncContext *acc) {
  3282. if (!acc)
  3283. return;
  3284. redisClusterContext *cc = acc->cc;
  3285. acc->err = cc->err;
  3286. memcpy(acc->errstr, cc->errstr, 128);
  3287. }
  3288. static void __redisClusterAsyncSetError(redisClusterAsyncContext *acc,
  3289. int type, const char *str) {
  3290. size_t len;
  3291. acc->err = type;
  3292. if (str != NULL) {
  3293. len = strlen(str);
  3294. len = len < (sizeof(acc->errstr)-1) ? len : (sizeof(acc->errstr)-1);
  3295. memcpy(acc->errstr,str,len);
  3296. acc->errstr[len] = '\0';
  3297. } else {
  3298. /* Only REDIS_ERR_IO may lack a description! */
  3299. assert(type == REDIS_ERR_IO);
  3300. __redis_strerror_r(errno, acc->errstr, sizeof(acc->errstr));
  3301. }
  3302. }
  3303. static redisClusterAsyncContext *redisClusterAsyncInitialize(redisClusterContext *cc) {
  3304. redisClusterAsyncContext *acc;
  3305. if(cc == NULL)
  3306. {
  3307. return NULL;
  3308. }
  3309. acc = hi_alloc(sizeof(redisClusterAsyncContext));
  3310. if (acc == NULL)
  3311. return NULL;
  3312. acc->cc = cc;
  3313. acc->err = 0;
  3314. acc->data = NULL;
  3315. acc->adapter = NULL;
  3316. acc->attach_fn = NULL;
  3317. acc->onConnect = NULL;
  3318. acc->onDisconnect = NULL;
  3319. return acc;
  3320. }
  3321. static cluster_async_data *cluster_async_data_get(void)
  3322. {
  3323. cluster_async_data *cad;
  3324. cad = hi_alloc(sizeof(cluster_async_data));
  3325. if(cad == NULL)
  3326. {
  3327. return NULL;
  3328. }
  3329. cad->acc = NULL;
  3330. cad->command = NULL;
  3331. cad->callback = NULL;
  3332. cad->privdata = NULL;
  3333. cad->retry_count = 0;
  3334. return cad;
  3335. }
  3336. static void cluster_async_data_free(cluster_async_data *cad)
  3337. {
  3338. if(cad == NULL)
  3339. {
  3340. return;
  3341. }
  3342. if(cad->command != NULL)
  3343. {
  3344. command_destroy(cad->command);
  3345. }
  3346. hi_free(cad);
  3347. cad = NULL;
  3348. }
  3349. static void unlinkAsyncContextAndNode(redisAsyncContext* ac)
  3350. {
  3351. cluster_node *node;
  3352. if (ac->data) {
  3353. node = (cluster_node *)(ac->data);
  3354. node->acon = NULL;
  3355. }
  3356. }
  3357. redisAsyncContext * actx_get_by_node(redisClusterAsyncContext *acc,
  3358. cluster_node *node)
  3359. {
  3360. redisAsyncContext *ac;
  3361. if(node == NULL)
  3362. {
  3363. return NULL;
  3364. }
  3365. ac = node->acon;
  3366. if(ac != NULL)
  3367. {
  3368. if (ac->c.err == 0) {
  3369. return ac;
  3370. } else {
  3371. NOT_REACHED();
  3372. }
  3373. }
  3374. if(node->host == NULL || node->port <= 0)
  3375. {
  3376. __redisClusterAsyncSetError(acc, REDIS_ERR_OTHER, "node host or port is error");
  3377. return NULL;
  3378. }
  3379. ac = redisAsyncConnect(node->host, node->port);
  3380. if(ac == NULL)
  3381. {
  3382. __redisClusterAsyncSetError(acc, REDIS_ERR_OTHER, "node host or port is error");
  3383. return NULL;
  3384. }
  3385. if(acc->adapter)
  3386. {
  3387. acc->attach_fn(ac, acc->adapter);
  3388. }
  3389. if(acc->onConnect)
  3390. {
  3391. redisAsyncSetConnectCallback(ac, acc->onConnect);
  3392. }
  3393. if(acc->onDisconnect)
  3394. {
  3395. redisAsyncSetDisconnectCallback(ac, acc->onDisconnect);
  3396. }
  3397. ac->data = node;
  3398. ac->dataHandler = unlinkAsyncContextAndNode;
  3399. node->acon = ac;
  3400. return ac;
  3401. }
  3402. static redisAsyncContext *actx_get_after_update_route_by_slot(
  3403. redisClusterAsyncContext *acc, int slot_num)
  3404. {
  3405. int ret;
  3406. redisClusterContext *cc;
  3407. redisAsyncContext *ac;
  3408. cluster_node *node;
  3409. if(acc == NULL || slot_num < 0)
  3410. {
  3411. return NULL;
  3412. }
  3413. cc = acc->cc;
  3414. if(cc == NULL)
  3415. {
  3416. return NULL;
  3417. }
  3418. ret = cluster_update_route(cc);
  3419. if(ret != REDIS_OK)
  3420. {
  3421. __redisClusterAsyncSetError(acc, REDIS_ERR_OTHER,
  3422. "route update error, please recreate redisClusterContext!");
  3423. return NULL;
  3424. }
  3425. node = node_get_by_table(cc, (uint32_t)slot_num);
  3426. if(node == NULL)
  3427. {
  3428. __redisClusterAsyncSetError(acc,
  3429. REDIS_ERR_OTHER, "node get by table error");
  3430. return NULL;
  3431. }
  3432. ac = actx_get_by_node(acc, node);
  3433. if(ac == NULL)
  3434. {
  3435. __redisClusterAsyncSetError(acc,
  3436. REDIS_ERR_OTHER, "actx get by node error");
  3437. return NULL;
  3438. }
  3439. else if(ac->err)
  3440. {
  3441. __redisClusterAsyncSetError(acc, ac->err, ac->errstr);
  3442. return NULL;
  3443. }
  3444. return ac;
  3445. }
  3446. redisClusterAsyncContext *redisClusterAsyncConnect(const char *addrs, int flags) {
  3447. redisClusterContext *cc;
  3448. redisClusterAsyncContext *acc;
  3449. cc = redisClusterConnectNonBlock(addrs, flags);
  3450. if(cc == NULL)
  3451. {
  3452. return NULL;
  3453. }
  3454. acc = redisClusterAsyncInitialize(cc);
  3455. if (acc == NULL) {
  3456. redisClusterFree(cc);
  3457. return NULL;
  3458. }
  3459. __redisClusterAsyncCopyError(acc);
  3460. return acc;
  3461. }
  3462. int redisClusterAsyncSetConnectCallback(
  3463. redisClusterAsyncContext *acc, redisConnectCallback *fn)
  3464. {
  3465. if (acc->onConnect == NULL) {
  3466. acc->onConnect = fn;
  3467. return REDIS_OK;
  3468. }
  3469. return REDIS_ERR;
  3470. }
  3471. int redisClusterAsyncSetDisconnectCallback(
  3472. redisClusterAsyncContext *acc, redisDisconnectCallback *fn)
  3473. {
  3474. if (acc->onDisconnect == NULL) {
  3475. acc->onDisconnect = fn;
  3476. return REDIS_OK;
  3477. }
  3478. return REDIS_ERR;
  3479. }
  3480. static void redisClusterAsyncCallback(redisAsyncContext *ac, void *r, void *privdata) {
  3481. int ret;
  3482. redisReply *reply = r;
  3483. cluster_async_data *cad = privdata;
  3484. redisClusterAsyncContext *acc;
  3485. redisClusterContext *cc;
  3486. redisAsyncContext *ac_retry = NULL;
  3487. int error_type;
  3488. cluster_node *node;
  3489. struct cmd *command;
  3490. int64_t now, next;
  3491. if(cad == NULL)
  3492. {
  3493. goto error;
  3494. }
  3495. acc = cad->acc;
  3496. if(acc == NULL)
  3497. {
  3498. goto error;
  3499. }
  3500. cc = acc->cc;
  3501. if(cc == NULL)
  3502. {
  3503. goto error;
  3504. }
  3505. command = cad->command;
  3506. if(command == NULL)
  3507. {
  3508. goto error;
  3509. }
  3510. if(reply == NULL)
  3511. {
  3512. //Note:
  3513. //I can't decide witch is the best way to deal with connect
  3514. //problem for hiredis cluster async api.
  3515. //But now the way is : when enough null reply for a node,
  3516. //we will update the route after the cluster node timeout.
  3517. //If you have a better idea, please contact with me. Thank you.
  3518. //My email: [email protected]
  3519. node = (cluster_node *)(ac->data);
  3520. ASSERT(node != NULL);
  3521. __redisClusterAsyncSetError(acc,
  3522. ac->err, ac->errstr);
  3523. if(cc->update_route_time != 0)
  3524. {
  3525. now = hi_usec_now();
  3526. if(now >= cc->update_route_time)
  3527. {
  3528. ret = cluster_update_route(cc);
  3529. if(ret != REDIS_OK)
  3530. {
  3531. __redisClusterAsyncSetError(acc, REDIS_ERR_OTHER,
  3532. "route update error, please recreate redisClusterContext!");
  3533. }
  3534. cc->update_route_time = 0LL;
  3535. }
  3536. goto done;
  3537. }
  3538. node->failure_count ++;
  3539. if(node->failure_count > cc->max_redirect_count)
  3540. {
  3541. char *cluster_timeout_str;
  3542. int cluster_timeout_str_len;
  3543. int cluster_timeout;
  3544. node->failure_count = 0;
  3545. if(cc->update_route_time != 0)
  3546. {
  3547. goto done;
  3548. }
  3549. cluster_timeout_str = cluster_config_get(cc,
  3550. "cluster-node-timeout", &cluster_timeout_str_len);
  3551. if(cluster_timeout_str == NULL)
  3552. {
  3553. __redisClusterAsyncSetError(acc,
  3554. cc->err, cc->errstr);
  3555. goto done;
  3556. }
  3557. cluster_timeout = hi_atoi(cluster_timeout_str,
  3558. cluster_timeout_str_len);
  3559. free(cluster_timeout_str);
  3560. if(cluster_timeout <= 0)
  3561. {
  3562. __redisClusterAsyncSetError(acc,
  3563. REDIS_ERR_OTHER,
  3564. "cluster_timeout_str convert to integer error");
  3565. goto done;
  3566. }
  3567. now = hi_usec_now();
  3568. if (now < 0) {
  3569. __redisClusterAsyncSetError(acc,
  3570. REDIS_ERR_OTHER,
  3571. "get now usec time error");
  3572. goto done;
  3573. }
  3574. next = now + (cluster_timeout * 1000LL);
  3575. cc->update_route_time = next;
  3576. }
  3577. goto done;
  3578. }
  3579. error_type = cluster_reply_error_type(reply);
  3580. if(error_type > CLUSTER_NOT_ERR && error_type < CLUSTER_ERR_SENTINEL)
  3581. {
  3582. cad->retry_count ++;
  3583. if(cad->retry_count > cc->max_redirect_count)
  3584. {
  3585. cad->retry_count = 0;
  3586. __redisClusterAsyncSetError(acc,
  3587. REDIS_ERR_CLUSTER_TOO_MANY_REDIRECT,
  3588. "too many cluster redirect");
  3589. goto done;
  3590. }
  3591. switch(error_type)
  3592. {
  3593. case CLUSTER_ERR_MOVED:
  3594. ac_retry = actx_get_after_update_route_by_slot(acc, command->slot_num);
  3595. if(ac_retry == NULL)
  3596. {
  3597. goto done;
  3598. }
  3599. break;
  3600. case CLUSTER_ERR_ASK:
  3601. node = node_get_by_ask_error_reply(cc, reply);
  3602. if(node == NULL)
  3603. {
  3604. __redisClusterAsyncSetError(acc,
  3605. cc->err, cc->errstr);
  3606. goto done;
  3607. }
  3608. ac_retry = actx_get_by_node(acc, node);
  3609. if(ac_retry == NULL)
  3610. {
  3611. __redisClusterAsyncSetError(acc,
  3612. REDIS_ERR_OTHER, "actx get by node error");
  3613. goto done;
  3614. }
  3615. else if(ac_retry->err)
  3616. {
  3617. __redisClusterAsyncSetError(acc,
  3618. ac_retry->err, ac_retry->errstr);
  3619. goto done;
  3620. }
  3621. ret = redisAsyncCommand(ac_retry,
  3622. NULL,NULL,REDIS_COMMAND_ASKING);
  3623. if(ret != REDIS_OK)
  3624. {
  3625. goto error;
  3626. }
  3627. break;
  3628. case CLUSTER_ERR_TRYAGAIN:
  3629. case CLUSTER_ERR_CROSSSLOT:
  3630. case CLUSTER_ERR_CLUSTERDOWN:
  3631. ac_retry = ac;
  3632. break;
  3633. default:
  3634. goto done;
  3635. break;
  3636. }
  3637. goto retry;
  3638. }
  3639. done:
  3640. if(acc->err)
  3641. {
  3642. cad->callback(acc, NULL, cad->privdata);
  3643. }
  3644. else
  3645. {
  3646. cad->callback(acc, r, cad->privdata);
  3647. }
  3648. if(cc->err)
  3649. {
  3650. cc->err = 0;
  3651. memset(cc->errstr, '\0', strlen(cc->errstr));
  3652. }
  3653. if(acc->err)
  3654. {
  3655. acc->err = 0;
  3656. memset(acc->errstr, '\0', strlen(acc->errstr));
  3657. }
  3658. if(cad != NULL)
  3659. {
  3660. cluster_async_data_free(cad);
  3661. }
  3662. return;
  3663. retry:
  3664. ret = redisAsyncFormattedCommand(ac_retry,
  3665. redisClusterAsyncCallback,cad,command->cmd,command->clen);
  3666. if(ret != REDIS_OK)
  3667. {
  3668. goto error;
  3669. }
  3670. return;
  3671. error:
  3672. if(cad != NULL)
  3673. {
  3674. cluster_async_data_free(cad);
  3675. }
  3676. }
  3677. int redisClusterAsyncFormattedCommand(redisClusterAsyncContext *acc,
  3678. redisClusterCallbackFn *fn, void *privdata, char *cmd, int len) {
  3679. redisClusterContext *cc;
  3680. int status = REDIS_OK;
  3681. int slot_num;
  3682. cluster_node *node;
  3683. redisAsyncContext *ac;
  3684. struct cmd *command = NULL;
  3685. hilist *commands = NULL;
  3686. cluster_async_data *cad;
  3687. if(acc == NULL)
  3688. {
  3689. return REDIS_ERR;
  3690. }
  3691. cc = acc->cc;
  3692. if(cc->err)
  3693. {
  3694. cc->err = 0;
  3695. memset(cc->errstr, '\0', strlen(cc->errstr));
  3696. }
  3697. if(acc->err)
  3698. {
  3699. acc->err = 0;
  3700. memset(acc->errstr, '\0', strlen(acc->errstr));
  3701. }
  3702. command = command_get();
  3703. if(command == NULL)
  3704. {
  3705. __redisClusterAsyncSetError(acc,REDIS_ERR_OOM,"Out of memory");
  3706. goto error;
  3707. }
  3708. command->cmd = malloc(len*sizeof(*command->cmd));
  3709. if(command->cmd == NULL)
  3710. {
  3711. __redisClusterAsyncSetError(acc,REDIS_ERR_OOM,"Out of memory");
  3712. goto error;
  3713. }
  3714. memcpy(command->cmd, cmd, len);
  3715. command->clen = len;
  3716. commands = listCreate();
  3717. if(commands == NULL)
  3718. {
  3719. __redisClusterAsyncSetError(acc,REDIS_ERR_OOM,"Out of memory");
  3720. goto error;
  3721. }
  3722. commands->free = listCommandFree;
  3723. slot_num = command_format_by_slot(cc, command, commands);
  3724. if(slot_num < 0)
  3725. {
  3726. __redisClusterAsyncSetError(acc,
  3727. cc->err, cc->errstr);
  3728. goto error;
  3729. }
  3730. else if(slot_num >= REDIS_CLUSTER_SLOTS)
  3731. {
  3732. __redisClusterAsyncSetError(acc,
  3733. REDIS_ERR_OTHER,"slot_num is out of range");
  3734. goto error;
  3735. }
  3736. //all keys not belong to one slot
  3737. if(listLength(commands) > 0)
  3738. {
  3739. ASSERT(listLength(commands) != 1);
  3740. __redisClusterAsyncSetError(acc,REDIS_ERR_OTHER,
  3741. "Asynchronous API now not support multi-key command");
  3742. goto error;
  3743. }
  3744. node = node_get_by_table(cc, (uint32_t) slot_num);
  3745. if(node == NULL)
  3746. {
  3747. __redisClusterAsyncSetError(acc,
  3748. REDIS_ERR_OTHER, "node get by table error");
  3749. goto error;
  3750. }
  3751. ac = actx_get_by_node(acc, node);
  3752. if(ac == NULL)
  3753. {
  3754. __redisClusterAsyncSetError(acc,
  3755. REDIS_ERR_OTHER, "actx get by node error");
  3756. goto error;
  3757. }
  3758. else if(ac->err)
  3759. {
  3760. __redisClusterAsyncSetError(acc, ac->err, ac->errstr);
  3761. goto error;
  3762. }
  3763. cad = cluster_async_data_get();
  3764. if(cad == NULL)
  3765. {
  3766. __redisClusterAsyncSetError(acc,REDIS_ERR_OOM,"Out of memory");
  3767. goto error;
  3768. }
  3769. cad->acc = acc;
  3770. cad->command = command;
  3771. cad->callback = fn;
  3772. cad->privdata = privdata;
  3773. status = redisAsyncFormattedCommand(ac,
  3774. redisClusterAsyncCallback,cad,cmd,len);
  3775. if(status != REDIS_OK)
  3776. {
  3777. goto error;
  3778. }
  3779. if(commands != NULL)
  3780. {
  3781. listRelease(commands);
  3782. }
  3783. return REDIS_OK;
  3784. error:
  3785. if(command != NULL)
  3786. {
  3787. command_destroy(command);
  3788. }
  3789. if(commands != NULL)
  3790. {
  3791. listRelease(commands);
  3792. }
  3793. return REDIS_ERR;
  3794. }
  3795. int redisClustervAsyncCommand(redisClusterAsyncContext *acc,
  3796. redisClusterCallbackFn *fn, void *privdata, const char *format, va_list ap) {
  3797. int ret;
  3798. char *cmd;
  3799. int len;
  3800. if(acc == NULL)
  3801. {
  3802. return REDIS_ERR;
  3803. }
  3804. len = redisvFormatCommand(&cmd,format,ap);
  3805. if (len == -1) {
  3806. __redisClusterAsyncSetError(acc,REDIS_ERR_OOM,"Out of memory");
  3807. return REDIS_ERR;
  3808. } else if (len == -2) {
  3809. __redisClusterAsyncSetError(acc,REDIS_ERR_OTHER,"Invalid format string");
  3810. return REDIS_ERR;
  3811. }
  3812. ret = redisClusterAsyncFormattedCommand(acc, fn, privdata, cmd, len);
  3813. free(cmd);
  3814. return ret;
  3815. }
  3816. int redisClusterAsyncCommand(redisClusterAsyncContext *acc,
  3817. redisClusterCallbackFn *fn, void *privdata, const char *format, ...) {
  3818. int ret;
  3819. va_list ap;
  3820. va_start(ap,format);
  3821. ret = redisClustervAsyncCommand(acc, fn, privdata, format, ap);
  3822. va_end(ap);
  3823. return ret;
  3824. }
  3825. int redisClusterAsyncCommandArgv(redisClusterAsyncContext *acc,
  3826. redisClusterCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen) {
  3827. int ret;
  3828. char *cmd;
  3829. int len;
  3830. len = redisFormatCommandArgv(&cmd,argc,argv,argvlen);
  3831. if (len == -1) {
  3832. __redisClusterAsyncSetError(acc,REDIS_ERR_OOM,"Out of memory");
  3833. return REDIS_ERR;
  3834. }
  3835. ret = redisClusterAsyncFormattedCommand(acc, fn, privdata, cmd, len);
  3836. free(cmd);
  3837. return ret;
  3838. }
  3839. void redisClusterAsyncDisconnect(redisClusterAsyncContext *acc) {
  3840. redisClusterContext *cc;
  3841. redisAsyncContext *ac;
  3842. dictIterator *di;
  3843. dictEntry *de;
  3844. dict *nodes;
  3845. struct cluster_node *node;
  3846. if(acc == NULL)
  3847. {
  3848. return;
  3849. }
  3850. cc = acc->cc;
  3851. nodes = cc->nodes;
  3852. if(nodes == NULL)
  3853. {
  3854. return;
  3855. }
  3856. di = dictGetIterator(nodes);
  3857. while((de = dictNext(di)) != NULL)
  3858. {
  3859. node = dictGetEntryVal(de);
  3860. ac = node->acon;
  3861. if(ac == NULL || ac->err)
  3862. {
  3863. continue;
  3864. }
  3865. redisAsyncDisconnect(ac);
  3866. node->acon = NULL;
  3867. }
  3868. }
  3869. void redisClusterAsyncFree(redisClusterAsyncContext *acc)
  3870. {
  3871. redisClusterContext *cc;
  3872. if(acc == NULL)
  3873. {
  3874. return;
  3875. }
  3876. cc = acc->cc;
  3877. redisClusterFree(cc);
  3878. hi_free(acc);
  3879. }