mirror/ULib
1
0
Fork 0
mirror of https://github.com/stefanocasazza/ULib.git synced 2025-09-28 19:05:55 +08:00
Code Issues Releases Wiki Activity

sync

Browse Source
This commit is contained in:
stefanocasazza 2019-03-08 15:59:40 +01:00
parent c233caf07a
commit 8976b94e73
11 changed files with 439 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/ulib/base/utility.h
View File

@ -261,6 +261,7 @@ U_EXPORT char* u_memoryDump( char* restrict bp, unsigned char* restrict cp, u
U_EXPORT uint32_t u_memory_dump(char* restrict bp, unsigned char* restrict cp, uint32_t n);
U_EXPORT uint8_t u_get_loadavg(void); /* Get the load average of the system (over last 1 minute) */
U_EXPORT uint16_t u_crc16(const char* a, uint32_t len); /* CRC16 implementation according to CCITT standards */
U_EXPORT uint32_t u_printSize(char* restrict buffer, uint64_t bytes); /* print size using u_calcRate() */
U_EXPORT int u_getScreenWidth(void) __pure; /* Determine the width of the terminal we're running on */

2
include/ulib/net/client/client.h
View File

@ -316,7 +316,7 @@ protected:
~UClient_Base();
private:
U_DISALLOW_COPY_AND_ASSIGN(UClient_Base)
// U_DISALLOW_COPY_AND_ASSIGN(UClient_Base)
static USocket* csocket;
static vPFu resize_response_buffer;

153
include/ulib/net/client/redis.h
View File

@ -15,6 +15,7 @@
#define ULIB_REDIS_H 1
#include <ulib/notifier.h>
#include <ulib/net/tcpsocket.h>
#include <ulib/net/unixsocket.h>
#include <ulib/net/client/client.h>
@ -62,6 +63,8 @@
typedef void (*vPFcs) (const UString&);
typedef void (*vPFcscs)(const UString&,const UString&);
class UREDISClusterClient;
class U_EXPORT UREDISClient_Base : public UClient_Base, UEventFd {
public:
@ -189,6 +192,45 @@ public:
bool connect(const char* host = U_NULLPTR, unsigned int _port = 6379);
// by Victor Stewart
UString single(const UString& pipeline)
{
U_TRACE(0, "UREDISClient_Base::single(%V)", pipeline.rep)
(void) processRequest(U_RC_MULTIBULK, U_STRING_TO_PARAM(pipeline));
return vitem[0];
}
void silencedSingle(UString& pipeline)
{
U_TRACE(0, "UREDISClient_Base::silencedSingle(%V)", pipeline.rep)
(void) pipeline.insert(0, U_CONSTANT_TO_PARAM("CLIENT REPLY SKIP \r\n"));
(void) processRequest(U_RC_MULTIBULK, U_STRING_TO_PARAM(pipeline));
}
const UVector<UString>& multi(const UString& pipeline)
{
U_TRACE(0, "UREDISClient_Base::multi(%V)", pipeline.rep)
(void) processRequest(U_RC_MULTIBULK, U_STRING_TO_PARAM(pipeline));
return vitem;
}
void silencedMulti(UString& pipeline)
{
U_TRACE(0, "UREDISClient_Base::silencedMulti(%V)", pipeline.rep)
(void) pipeline.insert(0, U_CONSTANT_TO_PARAM("CLIENT REPLY OFF \r\n"));
(void) pipeline.append(U_CONSTANT_TO_PARAM("CLIENT REPLY ON \r\n"));
(void) processRequest(U_RC_MULTIBULK, U_STRING_TO_PARAM(pipeline));
}
// STRING (@see http://redis.io/commands#string)
bool get(const char* key, uint32_t keylen) // Get the value of a key
@ -857,7 +899,9 @@ protected:
private:
bool getResponseItem() U_NO_EXPORT;
U_DISALLOW_COPY_AND_ASSIGN(UREDISClient_Base)
friend class UREDISClusterClient;
// U_DISALLOW_COPY_AND_ASSIGN(UREDISClient_Base)
};
template <class Socket> class U_EXPORT UREDISClient : public UREDISClient_Base {
@ -882,7 +926,7 @@ public:
#endif
private:
U_DISALLOW_COPY_AND_ASSIGN(UREDISClient)
// U_DISALLOW_COPY_AND_ASSIGN(UREDISClient)
};
template <> class U_EXPORT UREDISClient<UUnixSocket> : public UREDISClient_Base {
@ -931,4 +975,109 @@ public:
private:
U_DISALLOW_COPY_AND_ASSIGN(UREDISClient<UUnixSocket>)
};
// by Victor Stewart
#if defined(U_STDCPP_ENABLE) && defined(HAVE_CXX17)
# include <vector>
class U_EXPORT UREDISClusterClient : public UREDISClient<UTCPSocket> {
private:
struct RedisNode {
UString ipAddress;
UREDISClient<UTCPSocket> client;
uint16_t lowHashSlot, highHashSlot;
};
enum class ClusterError : uint8_t {
none,
moved,
ask,
tryagain
};
ClusterError error;
UString temporaryASKip;
std::vector<RedisNode> redisNodes;
uint16_t hashslotForKey(const UString& hashableKey) { return u_crc16(U_STRING_TO_PARAM(hashableKey)); }
uint16_t hashslotFromCommand(const UString& command)
{
U_TRACE(0, "UREDISClusterClient::hashslotFromCommand(%V)", command.rep)
// expects hashable keys to be delivered as abc{hashableKey}xyz value blah \r\n
uint32_t beginning = command.find('{') + 1,
end = command.find('}', beginning) - 1;
return hashslotForKey(command.substr(beginning, end - beginning));
}
UREDISClient<UTCPSocket>& clientForHashslot(uint16_t hashslot)
{
U_TRACE(0, "UREDISClusterClient::clientForHashslot(%u)", hashslot)
for (RedisNode& workingNode : redisNodes)
{
if ((workingNode.lowHashSlot <= hashslot) || (workingNode.highHashSlot >= hashslot)) return workingNode.client;
}
return redisNodes[0].client;
}
UREDISClient<UTCPSocket>& clientForASKip()
{
for (RedisNode& workingNode : redisNodes)
{
if (temporaryASKip == workingNode.ipAddress) return workingNode.client;
}
return redisNodes[0].client;
}
UREDISClient<UTCPSocket>& clientForHashableKey(const UString& hashableKey) { return clientForHashslot(hashslotForKey(hashableKey)); }
public:
UREDISClusterClient() : UREDISClient<UTCPSocket>()
{
U_TRACE_CTOR(0, UREDISClusterClient, "")
}
~UREDISClusterClient()
{
U_TRACE_DTOR(0, UREDISClusterClient)
}
void processResponse();
void calculateNodeMap();
const UVector<UString>& processPipeline(UString& pipeline, bool silence);
// all of these multis require all keys to exist within a single hash slot (on the same node isn't good enough)
UString clusterSingle(const UString& hashableKey, const UString& pipeline) { return clientForHashableKey(hashableKey).single(pipeline); }
const UVector<UString>& clusterMulti( const UString& hashableKey, const UString& pipeline) { return clientForHashableKey(hashableKey).multi(pipeline); }
void clusterSilencedMulti( const UString& hashableKey, UString& pipeline) { clientForHashableKey(hashableKey).silencedMulti(pipeline); }
void clusterSilencedSingle(const UString& hashableKey, UString& pipeline) { clientForHashableKey(hashableKey).silencedSingle(pipeline); }
// anon multis are pipelined commands of various keys that might belong to many nodes. always processed in order. commands always delimined by \r\n
const UVector<UString>& clusterAnonMulti( UString& pipeline) { return processPipeline(pipeline, false); }
void clusterSilencedAnonMulti(UString& pipeline) { (void) processPipeline(pipeline, true); }
bool clusterUnsubscribe(const UString& hashableKey, const UString& channel) { return clientForHashableKey(hashableKey).unsubscribe(channel); }
bool clusterSubscribe( const UString& hashableKey, const UString& channel, vPFcscs callback) { return clientForHashableKey(hashableKey).subscribe(channel, callback); }
// DEBUG
#if defined(U_STDCPP_ENABLE) && defined(DEBUG)
const char* dump(bool _reset) const { return UREDISClient_Base::dump(_reset); }
#endif
private:
U_DISALLOW_COPY_AND_ASSIGN(UREDISClusterClient)
};
#endif
#endif

17
include/ulib/string.h
View File

@ -566,6 +566,22 @@ public:
// EXTENSION
bool isNumber(uint32_t pos) const
{
U_TRACE(0, "UStringRep::isNumber(%u)", pos)
U_CHECK_MEMORY
if (_length)
{
U_INTERNAL_ASSERT_MINOR(pos, _length)
if (u_isNumber(str + pos, _length - pos)) U_RETURN(true);
}
U_RETURN(false);
}
bool isBinary(uint32_t pos) const
{
U_TRACE(0, "UStringRep::isBinary(%u)", pos)
@ -2098,6 +2114,7 @@ public:
bool isText(uint32_t pos = 0) const { return rep->isText(pos); }
bool isUTF8(uint32_t pos = 0) const { return rep->isUTF8(pos); }
bool isUTF16(uint32_t pos = 0) const { return rep->isUTF16(pos); }
bool isNumber(uint32_t pos = 0) const { return rep->isNumber(pos); }
bool isBinary(uint32_t pos = 0) const { return rep->isBinary(pos); }
bool isBase64(uint32_t pos = 0) const { return rep->isBase64(pos); }
bool isBase64Url(uint32_t pos = 0) const { return rep->isBase64Url(pos); }

64
src/ulib/base/utility.c
View File

@ -821,6 +821,68 @@ __pure uint32_t u_findEndHeader(const char* restrict str, uint32_t n)
return endHeader;
}
/**
* CRC16 implementation according to CCITT standards
*
* Note by @antirez: this is actually the XMODEM CRC 16 algorithm, using the following parameters:
*
* Name : "XMODEM", also known as "ZMODEM", "CRC-16/ACORN"
* Width : 16 bit
* Poly : 1021 (That is actually x^16 + x^12 + x^5 + 1)
* Initialization : 0000
* Reflect Input byte : False
* Reflect Output CRC : False
* Xor constant to output CRC : 0000
* Output for "123456789" : 31C3
*/
uint16_t u_crc16(const char* buf, uint32_t len)
{
static uint16_t crc16tab[256]= {
0x0000,0x1021,0x2042,0x3063,0x4084,0x50a5,0x60c6,0x70e7,
0x8108,0x9129,0xa14a,0xb16b,0xc18c,0xd1ad,0xe1ce,0xf1ef,
0x1231,0x0210,0x3273,0x2252,0x52b5,0x4294,0x72f7,0x62d6,
0x9339,0x8318,0xb37b,0xa35a,0xd3bd,0xc39c,0xf3ff,0xe3de,
0x2462,0x3443,0x0420,0x1401,0x64e6,0x74c7,0x44a4,0x5485,
0xa56a,0xb54b,0x8528,0x9509,0xe5ee,0xf5cf,0xc5ac,0xd58d,
0x3653,0x2672,0x1611,0x0630,0x76d7,0x66f6,0x5695,0x46b4,
0xb75b,0xa77a,0x9719,0x8738,0xf7df,0xe7fe,0xd79d,0xc7bc,
0x48c4,0x58e5,0x6886,0x78a7,0x0840,0x1861,0x2802,0x3823,
0xc9cc,0xd9ed,0xe98e,0xf9af,0x8948,0x9969,0xa90a,0xb92b,
0x5af5,0x4ad4,0x7ab7,0x6a96,0x1a71,0x0a50,0x3a33,0x2a12,
0xdbfd,0xcbdc,0xfbbf,0xeb9e,0x9b79,0x8b58,0xbb3b,0xab1a,
0x6ca6,0x7c87,0x4ce4,0x5cc5,0x2c22,0x3c03,0x0c60,0x1c41,
0xedae,0xfd8f,0xcdec,0xddcd,0xad2a,0xbd0b,0x8d68,0x9d49,
0x7e97,0x6eb6,0x5ed5,0x4ef4,0x3e13,0x2e32,0x1e51,0x0e70,
0xff9f,0xefbe,0xdfdd,0xcffc,0xbf1b,0xaf3a,0x9f59,0x8f78,
0x9188,0x81a9,0xb1ca,0xa1eb,0xd10c,0xc12d,0xf14e,0xe16f,
0x1080,0x00a1,0x30c2,0x20e3,0x5004,0x4025,0x7046,0x6067,
0x83b9,0x9398,0xa3fb,0xb3da,0xc33d,0xd31c,0xe37f,0xf35e,
0x02b1,0x1290,0x22f3,0x32d2,0x4235,0x5214,0x6277,0x7256,
0xb5ea,0xa5cb,0x95a8,0x8589,0xf56e,0xe54f,0xd52c,0xc50d,
0x34e2,0x24c3,0x14a0,0x0481,0x7466,0x6447,0x5424,0x4405,
0xa7db,0xb7fa,0x8799,0x97b8,0xe75f,0xf77e,0xc71d,0xd73c,
0x26d3,0x36f2,0x0691,0x16b0,0x6657,0x7676,0x4615,0x5634,
0xd94c,0xc96d,0xf90e,0xe92f,0x99c8,0x89e9,0xb98a,0xa9ab,
0x5844,0x4865,0x7806,0x6827,0x18c0,0x08e1,0x3882,0x28a3,
0xcb7d,0xdb5c,0xeb3f,0xfb1e,0x8bf9,0x9bd8,0xabbb,0xbb9a,
0x4a75,0x5a54,0x6a37,0x7a16,0x0af1,0x1ad0,0x2ab3,0x3a92,
0xfd2e,0xed0f,0xdd6c,0xcd4d,0xbdaa,0xad8b,0x9de8,0x8dc9,
0x7c26,0x6c07,0x5c64,0x4c45,0x3ca2,0x2c83,0x1ce0,0x0cc1,
0xef1f,0xff3e,0xcf5d,0xdf7c,0xaf9b,0xbfba,0x8fd9,0x9ff8,
0x6e17,0x7e36,0x4e55,0x5e74,0x2e93,0x3eb2,0x0ed1,0x1ef0
};
uint16_t crc = 0;
uint32_t counter;
U_INTERNAL_TRACE("u_crc16(%.*s,%u)", U_min(len,128), buf, len)
for (counter = 0; counter < len; ++counter) crc = (crc<<8) ^ crc16tab[((crc>>8) ^ *buf++) & 0x00FF];
return crc;
}
/* Determine the width of the terminal we're running on */
__pure int u_getScreenWidth(void)
@ -1456,7 +1518,7 @@ __pure bool u_isNumber(const char* restrict s, uint32_t n)
((*(const unsigned char* restrict)s) >> 4) == 0x03 &&
vdigit[(*(const unsigned char* restrict)s) & 0x0f])
{
U_INTERNAL_PRINT("*s = %c, *s >> 4 = %c ", *s, (*(char* restrict)s) >> 4)
U_INTERNAL_PRINT("*s = %c, *s >> 4 = %c", *s, (*(char* restrict)s) >> 4)
++s;
}

195
src/ulib/net/client/redis.cpp
View File

@ -617,9 +617,201 @@ int UREDISClient_Base::handlerRead()
U_RETURN(U_NOTIFIER_OK);
}
#if defined(U_STDCPP_ENABLE)
// by Victor Stewart
# if defined(HAVE_CXX17)
void UREDISClusterClient::processResponse()
{
U_TRACE_NO_PARAM(0, "UREDISClusterClient::processResponse()")
if (UClient_Base::response.find("MOVED", 0, 5) != U_NOT_FOUND)
{
// MOVED 3999 127.0.0.1:6381 => the hashslot has been moved to another master node
error = ClusterError::moved;
calculateNodeMap();
}
else if (UClient_Base::response.find("ASK", 0, 3) != U_NOT_FOUND)
{
// ASK 3999 127.0.0.1:6381 => this means that one of the hash slots is being migrated to another server
error = ClusterError::ask;
uint32_t _start = UClient_Base::response.find(' ', 8) + 1,
end = UClient_Base::response.find(':', _start);
(void) temporaryASKip.assign(UClient_Base::response.substr(_start, end - _start));
}
else if (UClient_Base::response.find("TRYAGAIN", 0, 8) != U_NOT_FOUND)
{
/**
* during a resharding the multi-key operations targeting keys that all exist and are all still in the same node (either the source or destination node) are still available.
* Operations on keys that don't exist or are - during the resharding - split between the source and destination nodes, will generate a -TRYAGAIN error. The client can try
* the operation after some time, or report back the error. As soon as migration of the specified hash slot has terminated, all multi-key operations are available again for
* that hash slot
*/
error = ClusterError::tryagain;
UTimeVal(0L, 1000L).nanosleep(); // 0 sec, 1000 microsec = 1ms
}
else
{
error = ClusterError::none;
UREDISClient<UTCPSocket>::processResponse();
}
}
const UVector<UString>& UREDISClusterClient::processPipeline(UString& pipeline, bool silence)
{
U_TRACE(0, "UREDISClusterClient::processPipeline(%V,%b)", pipeline.rep, silence)
uint16_t hashslot = 0, workingHashslot;
UString command, workingString(U_CAPACITY);
UVector<UString> commands(pipeline, "\r\n");
for (uint32_t count = 0, index = 0, n = commands.size(); index < n; ++index)
{
command = commands[index];
workingHashslot = hashslotFromCommand(command);
if (workingHashslot == hashslot)
{
(void) workingString.append(command + "\r\n");
++count;
if ((index + 1) < n) continue;
}
hashslot = workingHashslot;
if (silence)
{
if (count > 1)
{
(void) workingString.insert(0, U_CONSTANT_TO_PARAM("CLIENT REPLY OFF \r\n"));
(void) workingString.append(U_CONSTANT_TO_PARAM("CLIENT REPLY ON \r\n"));
}
else
{
(void) pipeline.insert(0, U_CONSTANT_TO_PARAM("CLIENT REPLY SKIP \r\n"));
}
}
UREDISClient<UTCPSocket>& client = clientForHashslot(hashslot);
replay:
(void) client.processRequest(U_RC_MULTIBULK, U_STRING_TO_PARAM(workingString));
switch (error)
{
case ClusterError::moved:
case ClusterError::tryagain:
{
goto replay;
}
break;
case ClusterError::ask:
{
UREDISClient<UTCPSocket>& temporaryClient = clientForASKip();
(void) temporaryClient.processRequest(U_RC_MULTIBULK, U_STRING_TO_PARAM(workingString));
}
break;
case ClusterError::none: break;
}
if (silence == false) vitem.move(client.vitem);
}
return vitem;
}
void UREDISClusterClient::calculateNodeMap()
{
U_TRACE_NO_PARAM(0, "UREDISClusterClient::calculateNodeMap()")
/*
127.0.0.1:30001> cluster slots
1) 1) (integer) 0
2) (integer) 5460
3) 1) "127.0.0.1"
2) (integer) 30001
3) "09dbe9720cda62f7865eabc5fd8857c5d2678366"
4) 1) "127.0.0.1"
2) (integer) 30004
3) "821d8ca00d7ccf931ed3ffc7e3db0599d2271abf"
2) 1) (integer) 5461
2) (integer) 10922
3) 1) "127.0.0.1"
2) (integer) 30002
3) "c9d93d9f2c0c524ff34cc11838c2003d8c29e013"
4) 1) "127.0.0.1"
2) (integer) 30005
3) "faadb3eb99009de4ab72ad6b6ed87634c7ee410f"
3) 1) (integer) 10923
2) (integer) 16383
3) 1) "127.0.0.1"
2) (integer) 30003
3) "044ec91f325b7595e76dbcb18cc688b6a5b434a1"
4) 1) "127.0.0.1"
2) (integer) 30006
3) "58e6e48d41228013e5d9c1c37c5060693925e97e"
*/
bool findHashSlots = true;
uint16_t workingLowHashSlot;
uint16_t workingHighHashSlot;
(void) UREDISClient_Base::processRequest(U_RC_MULTIBULK, U_CONSTANT_TO_PARAM("CLUSTER SLOTS"));
const UVector<UString>& rawNodes = UREDISClient_Base::vitem;
for (uint32_t a = 0, b = rawNodes.size(); a < b; ++a)
{
if (findHashSlots)
{
if (rawNodes[a].isNumber() &&
rawNodes[a+1].isNumber())
{
workingLowHashSlot = rawNodes[a++].strtoul();
workingHighHashSlot = rawNodes[a].strtoul();
findHashSlots = false;
}
}
else
{
// the immediate next after hash slot is the master
RedisNode workingNode;
workingNode.lowHashSlot = workingLowHashSlot;
workingNode.highHashSlot = workingHighHashSlot;
(void) workingNode.ipAddress.assign(rawNodes[a]);
workingNode.client.connect(workingNode.ipAddress.c_str(), rawNodes[++a].strtoul());
redisNodes.push_back(std::move(workingNode));
findHashSlots = true;
}
}
}
# endif
// DEBUG
#if defined(U_STDCPP_ENABLE) && defined(DEBUG)
# if defined(DEBUG)
const char* UREDISClient_Base::dump(bool _reset) const
{
UClient_Base::dump(false);
@ -637,4 +829,5 @@ const char* UREDISClient_Base::dump(bool _reset) const
return U_NULLPTR;
}
# endif
#endif

1
tests/ulib/err/application.err Normal file
View File

@ -0,0 +1 @@
../.function: line 259: ../../examples/application/application: No such file or directory

0
tests/ulib/out/application.out Normal file
View File

8
tests/ulib/plugin/.deps/product1.Plo
View File

@ -206,8 +206,8 @@ plugin/product1.lo: plugin/product1.cpp /usr/include/stdc-predef.h \
/usr/include/unicode/utf16.h /usr/include/unicode/utf_old.h \
/usr/include/unicode/uenum.h /usr/include/unicode/localpointer.h \
/usr/include/libxml2/libxml/xmlIO.h \
/usr/include/libxml2/libxml/globals.h /usr/include/libxml2/libxml/SAX.h \
/usr/include/libxml2/libxml/xlink.h /usr/include/libxml2/libxml/SAX2.h \
/usr/include/libxml2/libxml/globals.h /usr/include/libxml2/libxml/SAX2.h \
/usr/include/libxml2/libxml/xlink.h \
/usr/include/libxml2/libxml/xmlmemory.h \
/usr/include/libxml2/libxml/threads.h \
../../include/ulib/internal/macro.h \
@ -809,12 +809,10 @@ plugin/product.h:
/usr/include/libxml2/libxml/globals.h:
/usr/include/libxml2/libxml/SAX.h:
/usr/include/libxml2/libxml/SAX2.h:
/usr/include/libxml2/libxml/xlink.h:
/usr/include/libxml2/libxml/SAX2.h:
/usr/include/libxml2/libxml/xmlmemory.h:
/usr/include/libxml2/libxml/threads.h:

8
tests/ulib/plugin/.deps/product2.Plo
View File

@ -206,8 +206,8 @@ plugin/product2.lo: plugin/product2.cpp /usr/include/stdc-predef.h \
/usr/include/unicode/utf16.h /usr/include/unicode/utf_old.h \
/usr/include/unicode/uenum.h /usr/include/unicode/localpointer.h \
/usr/include/libxml2/libxml/xmlIO.h \
/usr/include/libxml2/libxml/globals.h /usr/include/libxml2/libxml/SAX.h \
/usr/include/libxml2/libxml/xlink.h /usr/include/libxml2/libxml/SAX2.h \
/usr/include/libxml2/libxml/globals.h /usr/include/libxml2/libxml/SAX2.h \
/usr/include/libxml2/libxml/xlink.h \
/usr/include/libxml2/libxml/xmlmemory.h \
/usr/include/libxml2/libxml/threads.h \
../../include/ulib/internal/macro.h \
@ -809,12 +809,10 @@ plugin/product.h:
/usr/include/libxml2/libxml/globals.h:
/usr/include/libxml2/libxml/SAX.h:
/usr/include/libxml2/libxml/SAX2.h:
/usr/include/libxml2/libxml/xlink.h:
/usr/include/libxml2/libxml/SAX2.h:
/usr/include/libxml2/libxml/xmlmemory.h:
/usr/include/libxml2/libxml/threads.h:

5
tests/ulib/tmp/c Normal file
View File

@ -0,0 +1,5 @@
c
c
c
c
c