From f2bde068c91f3819a0452f30df80bfd5f1b92449 Mon Sep 17 00:00:00 2001 From: MussoNero Date: Sun, 11 Oct 2020 19:11:38 +0200 Subject: [PATCH 1/9] Wip: fix connection goes silent in binance_websocket schedule connection retry if it fail, it will quit --- src/binance_websocket.cpp | 250 +++++++++++++++++++++++++++++--------- 1 file changed, 195 insertions(+), 55 deletions(-) diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index 09e7073..eb0a366 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -11,17 +11,72 @@ #include #include #include +#include using namespace binance; using namespace std; -static lws_context* context = NULL; +static struct lws_context *context; static map handles; - +static lws_sorted_usec_list_t _sul; static atomic lws_service_cancelled(0); +static void connect_client(lws_sorted_usec_list_t *sul); + +struct endpoint_connection { + lws_sorted_usec_list_t sul; /* schedule connection retry */ + struct lws *wsi; /* related wsi if any */ + uint16_t retry_count; /* count of consequetive retries */ + lws* conn; + CB callback_jason_func; + char* ws_path; +} endpoint_prop; + +/* + * The retry and backoff policy we want to use for our client connections + */ +static const uint32_t backoff_ms[] = { 1000, 1000*2, 1000*3, 1000*4, 1000*5, 1000*6, 1000*7, 1000*8, 1000*9, 1000*10}; + +static const lws_retry_bo_t retry = { + .retry_ms_table = backoff_ms, + .retry_ms_table_count = LWS_ARRAY_SIZE(backoff_ms), + .conceal_count = LWS_ARRAY_SIZE(backoff_ms), + .secs_since_valid_ping = 30, /* force PINGs after secs idle */ + .secs_since_valid_hangup = 100, /* hangup after secs idle */ + .jitter_percent = 15, + /* + * jitter_percent controls how much additional random delay is + * added to the actual interval to be used, defult 30 + */ +}; + +/* + * If we don't enable permessage-deflate ws extension, during times when there + * are many ws messages per second the server coalesces them inside a smaller + * number of larger ssl records, for >100 mps typically >2048 records. + * + * This is a problem, because the coalesced record cannot be send nor decrypted + * until the last part of the record is received, meaning additional latency + * for the earlier members of the coalesced record that have just been sitting + * there waiting for the last one to go out and be decrypted. + * + * permessage-deflate reduces the data size before the tls layer, for >100mps + * reducing the colesced records to ~1.2KB. + */ +static const struct lws_extension extensions[] = { + { + "permessage-deflate", + lws_extension_callback_pm_deflate, + "permessage-deflate" + "; client_no_context_takeover" + "; client_max_window_bits" + }, + { NULL, NULL, NULL /* terminator */ } +}; static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { + struct endpoint_connection *endpoint_prop = (struct endpoint_connection *)user; + switch (reason) { case LWS_CALLBACK_CLIENT_ESTABLISHED : @@ -35,7 +90,7 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void { string str_result = string(reinterpret_cast(in), len); Json::Reader reader; - Json::Value json_result; + Json::Value json_result; reader.parse(str_result , json_result); if (handles.find(wsi) != handles.end()) @@ -46,18 +101,21 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void Logger::write_log(" Error parsing incoming message : %s\n", e.what()); return 1; } - } + } break; case LWS_CALLBACK_CLIENT_WRITEABLE : break; case LWS_CALLBACK_CLOSED : - { + try{ if (handles.find(wsi) != handles.end()) handles.erase(wsi); - } - goto cancel; + }catch (exception &e) + { + Logger::write_log(" Error LWS_CALLBACK_CLOSED message : %s\n", e.what()); + } + goto do_retry; case LWS_CALLBACK_GET_THREAD_ID: { #ifdef __APPLE__ @@ -70,25 +128,59 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void #endif return (int)(uint64_t)tid; } - break; - case LWS_CALLBACK_CLIENT_CONNECTION_ERROR : - { - if (handles.find(wsi) != handles.end()) - handles.erase(wsi); - Logger::write_log(" LWS_CALLBACK_CLIENT_CONNECTION_ERROR\n"); - } - goto cancel; - default : - // Make compiler happy regarding unhandled enums. - break; + break; + + case LWS_CALLBACK_CLIENT_CONNECTION_ERROR : + lwsl_err("CLIENT_CONNECTION_ERROR: %s\n", + in ? (char *)in : "(null)"); + try{ + if (handles.find(wsi) != handles.end()) + handles.erase(wsi); + Logger::write_log(" LWS_CALLBACK_CLIENT_CONNECTION_ERROR\n"); + }catch (exception &e) + { + Logger::write_log(" Error LWS_CALLBACK_CLIENT_CONNECTION_ERROR message : %s\n", e.what()); + } + goto do_retry; + break; + + case LWS_CALLBACK_CLIENT_CLOSED: + /*lwsl_err("LWS_CALLBACK_CLIENT_CLOSED : %s\n", + in ? (char *)in : "(null)");*/ + try{ + if (handles.find(wsi) != handles.end()) + handles.erase(wsi); + }catch (exception &e) + { + Logger::write_log(" Error LWS_CALLBACK_CLIENT_CLOSED message : %s\n", e.what()); + } + goto do_retry; + break; + + default : + // Make compiler happy regarding unhandled enums. + break; } return 0; -cancel : +do_retry: + try{ + if (lws_retry_sul_schedule_retry_wsi(wsi, &endpoint_prop->sul, connect_client, + &endpoint_prop->retry_count)) + { + lwsl_err("%s: connection attempts exhausted\n", __func__); + atomic_store(&lws_service_cancelled, 1); + return -1; + } + }catch (exception &e) + { + Logger::write_log(" Error do_retry message : %s\n", e.what()); + atomic_store(&lws_service_cancelled, 1); + return -1; + } - atomic_store(&lws_service_cancelled, 1); - return -1; + return 0; } const lws_protocols protocols[] = @@ -99,63 +191,111 @@ const lws_protocols protocols[] = .per_session_data_size = 0, .rx_buffer_size = 65536, }, - + { NULL, NULL, 0, 0 } /* end */ }; -void binance::Websocket::init() +static void +sigint_handler(int sig) { - lws_context_creation_info info; - memset(&info, 0, sizeof(info)); + atomic_store(&lws_service_cancelled, 1); +} - info.port = CONTEXT_PORT_NO_LISTEN; - info.protocols = protocols; - info.gid = -1; - info.uid = -1; +/* + * Scheduled sul callback that starts the connection attempt + */ +static void connect_client(lws_sorted_usec_list_t *sul) +{ + struct endpoint_connection *endpoint_prop = lws_container_of(sul, struct endpoint_connection, sul); + struct lws_client_connect_info ccinfo; + + memset(&ccinfo, 0, sizeof(ccinfo)); + + ccinfo.context = context; + ccinfo.port = BINANCE_WS_PORT; + ccinfo.address = BINANCE_WS_HOST; + ccinfo.path = endpoint_prop->ws_path; + ccinfo.host = lws_canonical_hostname(context); + ccinfo.origin = "origin"; + ccinfo.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK; + ccinfo.protocol = protocols[0].name; + ccinfo.local_protocol_name = protocols[0].name; + ccinfo.retry_and_idle_policy = &retry; + ccinfo.userdata = endpoint_prop; + endpoint_prop->conn = lws_client_connect_via_info(&ccinfo); + if (!endpoint_prop->conn) + { + /* + * Failed... schedule a retry... we can't use the _retry_wsi() + * convenience wrapper api here because no valid wsi at this + * point. + */ + if (lws_retry_sul_schedule(context, 0, sul, &retry, + connect_client, &endpoint_prop->retry_count)) + { + lwsl_err("%s: connection attempts exhausted\n", __func__); + atomic_store(&lws_service_cancelled, 1); + } + handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; + }else{ + handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; + } +} + +void binance::Websocket::init() +{ + struct lws_context_creation_info info; + signal(SIGINT, sigint_handler); + memset(&info, 0, sizeof(info)); // This option is needed here to imply LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT // option, which must be set on newer versions of OpenSSL. info.options = LWS_SERVER_OPTION_REQUIRE_VALID_OPENSSL_CLIENT_CERT; + info.port = CONTEXT_PORT_NO_LISTEN; + info.gid = -1; + info.uid = -1; + info.protocols = protocols; + info.fd_limit_per_thread = 0; + info.extensions = extensions; context = lws_create_context(&info); + if (!context) { + lwsl_err("lws init failed\n"); + atomic_store(&lws_service_cancelled, 1); + return; + } else{ + atomic_store(&lws_service_cancelled, 0); + } } // Register call backs void binance::Websocket::connect_endpoint(CB cb, const char* path) { - char ws_path[1024]; - strcpy(ws_path, path); - - // Connect if we are not connected to the server. - lws_client_connect_info ccinfo = { 0 }; - ccinfo.context = context; - ccinfo.address = BINANCE_WS_HOST; - ccinfo.port = BINANCE_WS_PORT; - ccinfo.path = ws_path; - ccinfo.host = lws_canonical_hostname(context); - ccinfo.origin = "origin"; - ccinfo.protocol = protocols[0].name; - ccinfo.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK; - - lws* conn = lws_client_connect_via_info(&ccinfo); - handles[conn] = cb; + struct endpoint_connection *endpoint_prop = lws_container_of(&_sul, struct endpoint_connection, sul); + endpoint_prop->ws_path = const_cast(path); + endpoint_prop->callback_jason_func = cb; + connect_client(&_sul); + if (!lws_service_cancelled) + /* schedule the first client connection attempt to happen immediately */ + lws_sul_schedule(context, 0, &endpoint_prop->sul, connect_client, 1); } // Entering event loop void binance::Websocket::enter_event_loop(std::chrono::hours hours) { - auto start = std::chrono::steady_clock::now(); - auto end = start + hours; - do { - using namespace std::chrono; - - lws_service(context, 500); - if (lws_service_cancelled) - break; - } while (std::chrono::steady_clock::now() < end); + auto start = std::chrono::steady_clock::now(); + auto end = start + hours; + auto n = 0; + do { + n = lws_service(context, 10); + if (lws_service_cancelled) + { + lws_cancel_service(context); + break; + } + } while (n >= 0 && std::chrono::steady_clock::now() < end); atomic_store(&lws_service_cancelled, 0); lws_context_destroy(context); } - From 4f4600439cf1e56447c3dc4ff6a258cc72dacbf2 Mon Sep 17 00:00:00 2001 From: mussonero Date: Mon, 12 Oct 2020 12:46:17 +0200 Subject: [PATCH 2/9] Additional Fixes to binance_websocket. added additional notes and fixed "IF ELSE" statement + missing header --- src/binance_websocket.cpp | 113 +++++++++++++++++++++----------------- 1 file changed, 62 insertions(+), 51 deletions(-) diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index eb0a366..f855dc4 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -12,6 +12,7 @@ #include #include #include +#include using namespace binance; using namespace std; @@ -22,8 +23,12 @@ static lws_sorted_usec_list_t _sul; static atomic lws_service_cancelled(0); static void connect_client(lws_sorted_usec_list_t *sul); -struct endpoint_connection { - lws_sorted_usec_list_t sul; /* schedule connection retry */ +/* + * This "contains" the endpoint connection proprty and has + * the connection bound to it + */ +static struct endpoint_connection { + lws_sorted_usec_list_t sul; /* schedule connection retry */ struct lws *wsi; /* related wsi if any */ uint16_t retry_count; /* count of consequetive retries */ lws* conn; @@ -34,14 +39,20 @@ struct endpoint_connection { /* * The retry and backoff policy we want to use for our client connections */ -static const uint32_t backoff_ms[] = { 1000, 1000*2, 1000*3, 1000*4, 1000*5, 1000*6, 1000*7, 1000*8, 1000*9, 1000*10}; +static const uint32_t backoff_ms[] = { 1000, 1000*2, 1000*3, 1000*4, 1000*5}; +/* + * This struct sets the policy for delays between retries, + * and for how long a connection may be 'idle' + * before it first tries to ping / pong on it to confirm it's up, + * or drops the connection if still idle. + */ static const lws_retry_bo_t retry = { .retry_ms_table = backoff_ms, .retry_ms_table_count = LWS_ARRAY_SIZE(backoff_ms), - .conceal_count = LWS_ARRAY_SIZE(backoff_ms), - .secs_since_valid_ping = 30, /* force PINGs after secs idle */ - .secs_since_valid_hangup = 100, /* hangup after secs idle */ + .conceal_count = LWS_ARRAY_SIZE(backoff_ms)*2, + .secs_since_valid_ping = 30, /* force PINGs after secs idle */ + .secs_since_valid_hangup = 60, /* hangup after secs idle */ .jitter_percent = 15, /* * jitter_percent controls how much additional random delay is @@ -80,8 +91,10 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void switch (reason) { case LWS_CALLBACK_CLIENT_ESTABLISHED : - lws_callback_on_writable(wsi); - break; + lwsl_user("%s: established\n", __func__); + lws_callback_on_writable(wsi); + endpoint_prop->wsi = wsi; + break; case LWS_CALLBACK_CLIENT_RECEIVE : { @@ -107,15 +120,9 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void case LWS_CALLBACK_CLIENT_WRITEABLE : break; - case LWS_CALLBACK_CLOSED : - try{ - if (handles.find(wsi) != handles.end()) - handles.erase(wsi); - }catch (exception &e) - { - Logger::write_log(" Error LWS_CALLBACK_CLOSED message : %s\n", e.what()); - } - goto do_retry; + case LWS_CALLBACK_CLOSED : + goto do_retry; + case LWS_CALLBACK_GET_THREAD_ID: { #ifdef __APPLE__ @@ -133,29 +140,17 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void case LWS_CALLBACK_CLIENT_CONNECTION_ERROR : lwsl_err("CLIENT_CONNECTION_ERROR: %s\n", in ? (char *)in : "(null)"); - try{ - if (handles.find(wsi) != handles.end()) - handles.erase(wsi); - Logger::write_log(" LWS_CALLBACK_CLIENT_CONNECTION_ERROR\n"); - }catch (exception &e) - { - Logger::write_log(" Error LWS_CALLBACK_CLIENT_CONNECTION_ERROR message : %s\n", e.what()); - } - goto do_retry; + if (handles.find(wsi) != handles.end()) + handles.erase(wsi); + lws_cancel_service(lws_get_context(wsi)); + atomic_store(&lws_service_cancelled, 1); + return -1; break; case LWS_CALLBACK_CLIENT_CLOSED: - /*lwsl_err("LWS_CALLBACK_CLIENT_CLOSED : %s\n", - in ? (char *)in : "(null)");*/ - try{ - if (handles.find(wsi) != handles.end()) - handles.erase(wsi); - }catch (exception &e) - { - Logger::write_log(" Error LWS_CALLBACK_CLIENT_CLOSED message : %s\n", e.what()); - } + lwsl_err("CLIENT_CALLBACK_CLIENT_CLOSED: %s\n", + in ? (char *)in : ""); goto do_retry; - break; default : // Make compiler happy regarding unhandled enums. @@ -169,9 +164,12 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void if (lws_retry_sul_schedule_retry_wsi(wsi, &endpoint_prop->sul, connect_client, &endpoint_prop->retry_count)) { + if (handles.find(wsi) != handles.end()) + handles.erase(wsi); + lws_cancel_service(lws_get_context(wsi)); lwsl_err("%s: connection attempts exhausted\n", __func__); - atomic_store(&lws_service_cancelled, 1); - return -1; + atomic_store(&lws_service_cancelled, 0); + return 0; } }catch (exception &e) { @@ -183,21 +181,22 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void return 0; } -const lws_protocols protocols[] = -{ +static const lws_protocols protocols[] = { - .name = "binance-websocket-api", - .callback = event_cb, - .per_session_data_size = 0, - .rx_buffer_size = 65536, - }, + { + .name = "binance-websocket-api", + .callback = event_cb, + .per_session_data_size = 0, + .rx_buffer_size = 65536, + }, - { NULL, NULL, 0, 0 } /* end */ -}; + { NULL, NULL, 0, 0 } /* end */ + }; static void sigint_handler(int sig) { + Logger::write_log(" Interactive attention signal : %d\n", sig); atomic_store(&lws_service_cancelled, 1); } @@ -206,7 +205,7 @@ sigint_handler(int sig) */ static void connect_client(lws_sorted_usec_list_t *sul) { - struct endpoint_connection *endpoint_prop = lws_container_of(sul, struct endpoint_connection, sul); + struct endpoint_connection *endpoint_prop = lws_container_of(&_sul, struct endpoint_connection, sul); struct lws_client_connect_info ccinfo; memset(&ccinfo, 0, sizeof(ccinfo)); @@ -222,6 +221,12 @@ static void connect_client(lws_sorted_usec_list_t *sul) ccinfo.local_protocol_name = protocols[0].name; ccinfo.retry_and_idle_policy = &retry; ccinfo.userdata = endpoint_prop; + /* + * We store the new wsi here early in the connection process, + * this gives the callback a way to identify which wsi faced the error + * even before the new wsi is returned and even if ultimately no wsi is returned. + */ + ccinfo.pwsi = &endpoint_prop->wsi; endpoint_prop->conn = lws_client_connect_via_info(&ccinfo); if (!endpoint_prop->conn) @@ -236,8 +241,14 @@ static void connect_client(lws_sorted_usec_list_t *sul) { lwsl_err("%s: connection attempts exhausted\n", __func__); atomic_store(&lws_service_cancelled, 1); + assert(endpoint_prop->wsi); + if ((endpoint_prop->wsi) && handles.find(endpoint_prop->wsi) != handles.end()) + handles.erase(endpoint_prop->wsi); + return; + } + else{ + handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; } - handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; }else{ handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; } @@ -264,7 +275,7 @@ void binance::Websocket::init() atomic_store(&lws_service_cancelled, 1); return; } else{ - atomic_store(&lws_service_cancelled, 0); + atomic_store(&lws_service_cancelled, 0); } } @@ -287,7 +298,7 @@ void binance::Websocket::enter_event_loop(std::chrono::hours hours) auto end = start + hours; auto n = 0; do { - n = lws_service(context, 10); + n = lws_service(context, 1000); if (lws_service_cancelled) { lws_cancel_service(context); From 2851b5fbfc68bca359d963b12aa59821ad8e4c83 Mon Sep 17 00:00:00 2001 From: MussoNero Date: Wed, 14 Oct 2020 10:54:45 +0200 Subject: [PATCH 3/9] fix crashing on received signal in binance_websocket Segmentation fault can happen when we attempting to emit signals from within the static method. In other words, from the callback, we need to invoke some non-static method of WebSocket which would emit the needed signal --- src/binance_websocket.cpp | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index f855dc4..ba0b5f1 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -21,13 +21,13 @@ static struct lws_context *context; static map handles; static lws_sorted_usec_list_t _sul; static atomic lws_service_cancelled(0); -static void connect_client(lws_sorted_usec_list_t *sul); +void connect_client(lws_sorted_usec_list_t *sul); /* * This "contains" the endpoint connection proprty and has * the connection bound to it */ -static struct endpoint_connection { +struct endpoint_connection { lws_sorted_usec_list_t sul; /* schedule connection retry */ struct lws *wsi; /* related wsi if any */ uint16_t retry_count; /* count of consequetive retries */ @@ -39,7 +39,7 @@ static struct endpoint_connection { /* * The retry and backoff policy we want to use for our client connections */ -static const uint32_t backoff_ms[] = { 1000, 1000*2, 1000*3, 1000*4, 1000*5}; +const uint32_t backoff_ms[] = { 1000, 1000*2, 1000*3, 1000*4, 1000*5}; /* * This struct sets the policy for delays between retries, @@ -47,10 +47,10 @@ static const uint32_t backoff_ms[] = { 1000, 1000*2, 1000*3, 1000*4, 1000*5}; * before it first tries to ping / pong on it to confirm it's up, * or drops the connection if still idle. */ -static const lws_retry_bo_t retry = { +const lws_retry_bo_t retry = { .retry_ms_table = backoff_ms, .retry_ms_table_count = LWS_ARRAY_SIZE(backoff_ms), - .conceal_count = LWS_ARRAY_SIZE(backoff_ms)*2, + .conceal_count = LWS_ARRAY_SIZE(backoff_ms), .secs_since_valid_ping = 30, /* force PINGs after secs idle */ .secs_since_valid_hangup = 60, /* hangup after secs idle */ .jitter_percent = 15, @@ -73,7 +73,7 @@ static const lws_retry_bo_t retry = { * permessage-deflate reduces the data size before the tls layer, for >100mps * reducing the colesced records to ~1.2KB. */ -static const struct lws_extension extensions[] = { +const struct lws_extension extensions[] = { { "permessage-deflate", lws_extension_callback_pm_deflate, @@ -84,7 +84,7 @@ static const struct lws_extension extensions[] = { { NULL, NULL, NULL /* terminator */ } }; -static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) +int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { struct endpoint_connection *endpoint_prop = (struct endpoint_connection *)user; @@ -168,8 +168,8 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void handles.erase(wsi); lws_cancel_service(lws_get_context(wsi)); lwsl_err("%s: connection attempts exhausted\n", __func__); - atomic_store(&lws_service_cancelled, 0); - return 0; + atomic_store(&lws_service_cancelled, 1); + return -1; } }catch (exception &e) { @@ -181,7 +181,7 @@ static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void return 0; } -static const lws_protocols protocols[] = +const lws_protocols protocols[] = { { .name = "binance-websocket-api", @@ -193,7 +193,7 @@ static const lws_protocols protocols[] = { NULL, NULL, 0, 0 } /* end */ }; -static void +void sigint_handler(int sig) { Logger::write_log(" Interactive attention signal : %d\n", sig); @@ -203,7 +203,7 @@ sigint_handler(int sig) /* * Scheduled sul callback that starts the connection attempt */ -static void connect_client(lws_sorted_usec_list_t *sul) +void connect_client(lws_sorted_usec_list_t *sul) { struct endpoint_connection *endpoint_prop = lws_container_of(&_sul, struct endpoint_connection, sul); struct lws_client_connect_info ccinfo; From 3282989e53f19b0c6674ba176ab9055ad75bb413 Mon Sep 17 00:00:00 2001 From: MussoNero Date: Thu, 15 Oct 2020 10:06:12 +0200 Subject: [PATCH 4/9] Map and serialize every endpoint connection Note: a single connection can listen to a maximum of 1024 streams --- src/binance_websocket.cpp | 240 +++++++++++++++++++++++--------------- 1 file changed, 149 insertions(+), 91 deletions(-) diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index ba0b5f1..13a28e1 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -1,7 +1,7 @@ /* Author: tensaix2j Date : 2017/10/15 - + C++ library for Binance API. */ @@ -12,14 +12,11 @@ #include #include #include -#include using namespace binance; using namespace std; static struct lws_context *context; -static map handles; -static lws_sorted_usec_list_t _sul; static atomic lws_service_cancelled(0); void connect_client(lws_sorted_usec_list_t *sul); @@ -28,13 +25,17 @@ void connect_client(lws_sorted_usec_list_t *sul); * the connection bound to it */ struct endpoint_connection { - lws_sorted_usec_list_t sul; /* schedule connection retry */ + lws_sorted_usec_list_t _sul; /* schedule connection retry */ struct lws *wsi; /* related wsi if any */ uint16_t retry_count; /* count of consequetive retries */ lws* conn; - CB callback_jason_func; + CB json_cb; char* ws_path; -} endpoint_prop; +}; + +static std::map concurrent; +static std::map endpoints_prop; +static pthread_mutex_t lock_concurrent; /* serialize access */ /* * The retry and backoff policy we want to use for our client connections @@ -86,17 +87,24 @@ const struct lws_extension extensions[] = { int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { - struct endpoint_connection *endpoint_prop = (struct endpoint_connection *)user; + int m; switch (reason) { - case LWS_CALLBACK_CLIENT_ESTABLISHED : - lwsl_user("%s: established\n", __func__); - lws_callback_on_writable(wsi); - endpoint_prop->wsi = wsi; + case LWS_CALLBACK_CLIENT_ESTABLISHED : + for (std::pair n : concurrent) { + if (endpoints_prop[n.second].wsi == wsi) { + lws_callback_on_writable(wsi); + m = n.second; + endpoints_prop[m].wsi = wsi; + lwsl_user("%s: connection established with success concurrent:%d ws_path::%s\n", + __func__, n.second, endpoints_prop[n.second].ws_path); + break; + } + } break; - case LWS_CALLBACK_CLIENT_RECEIVE : + case LWS_CALLBACK_CLIENT_RECEIVE : { // Handle incomming messages here. try @@ -106,24 +114,40 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s Json::Value json_result; reader.parse(str_result , json_result); - if (handles.find(wsi) != handles.end()) - handles[wsi](json_result); + for (std::pair n : concurrent) { + if (endpoints_prop[n.second].wsi == wsi) { + m = n.second; + endpoints_prop[n.second].json_cb(json_result); + endpoints_prop[m].retry_count = 0; + lwsl_user("%s: incomming messages from %s\n", + __func__ , endpoints_prop[n.second].ws_path); + break; + } + } } catch (exception &e) { - Logger::write_log(" Error parsing incoming message : %s\n", e.what()); - return 1; + Logger::write_log(" Error parsing incoming message : %s\n", e.what()); + return 1; } } - break; + break; - case LWS_CALLBACK_CLIENT_WRITEABLE : - break; + case LWS_CALLBACK_CLIENT_WRITEABLE : + break; case LWS_CALLBACK_CLOSED : - goto do_retry; + lwsl_err("CALLBACK_CLOSED: %s\n", + in ? (char *)in : ""); + for (std::pair n : concurrent) { + if (endpoints_prop[n.second].wsi == wsi) { + m = n.second; + goto do_retry; + } + } + break; - case LWS_CALLBACK_GET_THREAD_ID: + case LWS_CALLBACK_GET_THREAD_ID: { #ifdef __APPLE__ // On OS X pthread_threadid_np() is used, as pthread_self() returns a structure. @@ -138,19 +162,28 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s break; case LWS_CALLBACK_CLIENT_CONNECTION_ERROR : - lwsl_err("CLIENT_CONNECTION_ERROR: %s\n", - in ? (char *)in : "(null)"); - if (handles.find(wsi) != handles.end()) - handles.erase(wsi); - lws_cancel_service(lws_get_context(wsi)); - atomic_store(&lws_service_cancelled, 1); - return -1; + for (std::pair n : concurrent) { + if (endpoints_prop[n.second].wsi == wsi) { + atomic_store(&lws_service_cancelled, 1); + endpoints_prop.erase(n.second); + concurrent.erase(n.second); + lwsl_err("CLIENT_CONNECTION_ERROR Unknown WIS: %s\n", + in ? (char *)in : "(null)"); + return -1; + } + } break; case LWS_CALLBACK_CLIENT_CLOSED: lwsl_err("CLIENT_CALLBACK_CLIENT_CLOSED: %s\n", in ? (char *)in : ""); - goto do_retry; + for (std::pair n : concurrent) { + if (endpoints_prop[n.second].wsi == wsi) { + m = n.second; + goto do_retry; + } + } + break; default : // Make compiler happy regarding unhandled enums. @@ -161,16 +194,25 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s do_retry: try{ - if (lws_retry_sul_schedule_retry_wsi(wsi, &endpoint_prop->sul, connect_client, - &endpoint_prop->retry_count)) + if (lws_retry_sul_schedule_retry_wsi(endpoints_prop[m].wsi, &endpoints_prop[m]._sul, connect_client, + &endpoints_prop[m].retry_count)) { - if (handles.find(wsi) != handles.end()) - handles.erase(wsi); - lws_cancel_service(lws_get_context(wsi)); - lwsl_err("%s: connection attempts exhausted\n", __func__); - atomic_store(&lws_service_cancelled, 1); - return -1; + if(endpoints_prop[m].retry_count > (LWS_ARRAY_SIZE(backoff_ms))){ + endpoints_prop.erase(m); + concurrent.erase(m); + atomic_store(&lws_service_cancelled, 1); + return -1; + } + { + lwsl_err("%s: connection attempts exhausted,we will keep retrying count:%d ws_path:%s\n", + __func__, endpoints_prop[m].retry_count, endpoints_prop[m].ws_path); + atomic_store(&lws_service_cancelled, 0); + lws_sul_schedule(lws_get_context(endpoints_prop[m].wsi), 0, &endpoints_prop[m]._sul, connect_client, 10 * LWS_US_PER_SEC); + return 0; + } } + lwsl_user("%s: connection attempts success, retrying count:%d ws_path:%s\n", + __func__, endpoints_prop[m].retry_count, endpoints_prop[m].ws_path); }catch (exception &e) { Logger::write_log(" Error do_retry message : %s\n", e.what()); @@ -205,53 +247,54 @@ sigint_handler(int sig) */ void connect_client(lws_sorted_usec_list_t *sul) { - struct endpoint_connection *endpoint_prop = lws_container_of(&_sul, struct endpoint_connection, sul); - struct lws_client_connect_info ccinfo; - - memset(&ccinfo, 0, sizeof(ccinfo)); - - ccinfo.context = context; - ccinfo.port = BINANCE_WS_PORT; - ccinfo.address = BINANCE_WS_HOST; - ccinfo.path = endpoint_prop->ws_path; - ccinfo.host = lws_canonical_hostname(context); - ccinfo.origin = "origin"; - ccinfo.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK; - ccinfo.protocol = protocols[0].name; - ccinfo.local_protocol_name = protocols[0].name; - ccinfo.retry_and_idle_policy = &retry; - ccinfo.userdata = endpoint_prop; - /* - * We store the new wsi here early in the connection process, - * this gives the callback a way to identify which wsi faced the error - * even before the new wsi is returned and even if ultimately no wsi is returned. - */ - ccinfo.pwsi = &endpoint_prop->wsi; - - endpoint_prop->conn = lws_client_connect_via_info(&ccinfo); - if (!endpoint_prop->conn) - { - /* - * Failed... schedule a retry... we can't use the _retry_wsi() - * convenience wrapper api here because no valid wsi at this - * point. - */ - if (lws_retry_sul_schedule(context, 0, sul, &retry, - connect_client, &endpoint_prop->retry_count)) - { - lwsl_err("%s: connection attempts exhausted\n", __func__); - atomic_store(&lws_service_cancelled, 1); - assert(endpoint_prop->wsi); - if ((endpoint_prop->wsi) && handles.find(endpoint_prop->wsi) != handles.end()) - handles.erase(endpoint_prop->wsi); - return; - } - else{ - handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; + for (std::pair n : concurrent) { + if (&endpoints_prop[n.second]._sul == sul) { + lwsl_user("%s: success ws_path::%s\n", + __func__, endpoints_prop[n.second].ws_path); + struct lws_client_connect_info ccinfo; + + memset(&ccinfo, 0, sizeof(ccinfo)); + + ccinfo.context = context; + ccinfo.port = BINANCE_WS_PORT; + ccinfo.address = BINANCE_WS_HOST; + ccinfo.path = endpoints_prop[n.second].ws_path; + ccinfo.host = lws_canonical_hostname(context); + ccinfo.origin = "origin"; + ccinfo.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK | LCCSCF_PIPELINE; + ccinfo.protocol = protocols[0].name; + ccinfo.local_protocol_name = protocols[0].name; + ccinfo.retry_and_idle_policy = &retry; + ccinfo.userdata = &endpoints_prop[n.second]; + /* + * We store the new wsi here early in the connection process, + * this gives the callback a way to identify which wsi faced the error + * even before the new wsi is returned and even if ultimately no wsi is returned. + */ + ccinfo.pwsi = &endpoints_prop[n.second].wsi; + + endpoints_prop[n.second].conn = lws_client_connect_via_info(&ccinfo); + if (!endpoints_prop[n.second].conn) + { + /* + * Failed... schedule a retry... we can't use the _retry_wsi() + * convenience wrapper api here because no valid wsi at this + * point. + */ + if (lws_retry_sul_schedule(context, 0, &endpoints_prop[n.second]._sul, &retry, + connect_client, &endpoints_prop[n.second].retry_count)) + { + lwsl_err("%s: connection attempts exhausted\n", __func__); + endpoints_prop.erase(n.second); + concurrent.erase(n.second); + atomic_store(&lws_service_cancelled, 1); + return; + } + } + break; } - }else{ - handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func; } + } void binance::Websocket::init() @@ -282,13 +325,27 @@ void binance::Websocket::init() // Register call backs void binance::Websocket::connect_endpoint(CB cb, const char* path) { - struct endpoint_connection *endpoint_prop = lws_container_of(&_sul, struct endpoint_connection, sul); - endpoint_prop->ws_path = const_cast(path); - endpoint_prop->callback_jason_func = cb; - connect_client(&_sul); + pthread_mutex_lock(&lock_concurrent); + if(concurrent.size() > 1024){ + lwsl_err("%s: maximum of 1024 connect_endpoints reached,\n", + __func__); + pthread_mutex_unlock(&lock_concurrent); + return; + } + int n = concurrent.size(); + concurrent.emplace(std::pair(n,n)); + endpoints_prop[n].ws_path = const_cast(path); + endpoints_prop[n].json_cb = cb; + connect_client(&endpoints_prop[n]._sul); + if (!lws_service_cancelled) - /* schedule the first client connection attempt to happen immediately */ - lws_sul_schedule(context, 0, &endpoint_prop->sul, connect_client, 1); + { + /* schedule the first client connection attempt to happen immediately */ + lws_sul_schedule(context, 0, &endpoints_prop[n]._sul, connect_client, 1); + lwsl_user("%s: concurrent:%d ws_path::%s\n", + __func__, n, endpoints_prop[n].ws_path); + } + pthread_mutex_unlock(&lock_concurrent); } // Entering event loop @@ -306,7 +363,8 @@ void binance::Websocket::enter_event_loop(std::chrono::hours hours) } } while (n >= 0 && std::chrono::steady_clock::now() < end); - atomic_store(&lws_service_cancelled, 0); + concurrent.clear(); + atomic_store(&lws_service_cancelled, 1); lws_context_destroy(context); -} +} \ No newline at end of file From 4c3e88a6690febe3299b489920d680b72266a8f0 Mon Sep 17 00:00:00 2001 From: mussonero Date: Sun, 18 Oct 2020 10:28:05 +0200 Subject: [PATCH 5/9] fix missing mutex_init & mutex_destroy fix missing pthread_mutex_init & pthread_mutex_destroy. --- src/binance_websocket.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index 13a28e1..c50d4da 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -299,6 +299,7 @@ void connect_client(lws_sorted_usec_list_t *sul) void binance::Websocket::init() { + pthread_mutex_init(&lock_concurrent, NULL); struct lws_context_creation_info info; signal(SIGINT, sigint_handler); memset(&info, 0, sizeof(info)); @@ -367,4 +368,5 @@ void binance::Websocket::enter_event_loop(std::chrono::hours hours) atomic_store(&lws_service_cancelled, 1); lws_context_destroy(context); -} \ No newline at end of file + pthread_mutex_destroy(&lock_concurrent); +} From 649627db02c994bf0105604d8543c51b660cfef6 Mon Sep 17 00:00:00 2001 From: MussoNero Date: Sun, 18 Oct 2020 16:13:07 +0200 Subject: [PATCH 6/9] remove unused lws_extesion & more mutex luck to event-callback remove unused lws_extesion & more mutex luck to event-callback --- src/binance_websocket.cpp | 44 ++++++++++++++++----------------------- 1 file changed, 18 insertions(+), 26 deletions(-) diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index c50d4da..c72df76 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -61,30 +61,6 @@ const lws_retry_bo_t retry = { */ }; -/* - * If we don't enable permessage-deflate ws extension, during times when there - * are many ws messages per second the server coalesces them inside a smaller - * number of larger ssl records, for >100 mps typically >2048 records. - * - * This is a problem, because the coalesced record cannot be send nor decrypted - * until the last part of the record is received, meaning additional latency - * for the earlier members of the coalesced record that have just been sitting - * there waiting for the last one to go out and be decrypted. - * - * permessage-deflate reduces the data size before the tls layer, for >100mps - * reducing the colesced records to ~1.2KB. - */ -const struct lws_extension extensions[] = { - { - "permessage-deflate", - lws_extension_callback_pm_deflate, - "permessage-deflate" - "; client_no_context_takeover" - "; client_max_window_bits" - }, - { NULL, NULL, NULL /* terminator */ } -}; - int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { int m; @@ -92,16 +68,18 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s switch (reason) { case LWS_CALLBACK_CLIENT_ESTABLISHED : + pthread_mutex_lock(&lock_concurrent); for (std::pair n : concurrent) { if (endpoints_prop[n.second].wsi == wsi) { lws_callback_on_writable(wsi); m = n.second; - endpoints_prop[m].wsi = wsi; lwsl_user("%s: connection established with success concurrent:%d ws_path::%s\n", __func__, n.second, endpoints_prop[n.second].ws_path); + pthread_mutex_unlock(&lock_concurrent); break; } } + pthread_mutex_unlock(&lock_concurrent); break; case LWS_CALLBACK_CLIENT_RECEIVE : @@ -109,6 +87,7 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s // Handle incomming messages here. try { + pthread_mutex_lock(&lock_concurrent); string str_result = string(reinterpret_cast(in), len); Json::Reader reader; Json::Value json_result; @@ -121,12 +100,15 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s endpoints_prop[m].retry_count = 0; lwsl_user("%s: incomming messages from %s\n", __func__ , endpoints_prop[n.second].ws_path); + pthread_mutex_unlock(&lock_concurrent); break; } } + pthread_mutex_unlock(&lock_concurrent); } catch (exception &e) { + pthread_mutex_unlock(&lock_concurrent); Logger::write_log(" Error parsing incoming message : %s\n", e.what()); return 1; } @@ -137,14 +119,17 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s break; case LWS_CALLBACK_CLOSED : + pthread_mutex_lock(&lock_concurrent); lwsl_err("CALLBACK_CLOSED: %s\n", in ? (char *)in : ""); for (std::pair n : concurrent) { if (endpoints_prop[n.second].wsi == wsi) { m = n.second; + pthread_mutex_unlock(&lock_concurrent); goto do_retry; } } + pthread_mutex_unlock(&lock_concurrent); break; case LWS_CALLBACK_GET_THREAD_ID: @@ -162,6 +147,7 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s break; case LWS_CALLBACK_CLIENT_CONNECTION_ERROR : + pthread_mutex_lock(&lock_concurrent); for (std::pair n : concurrent) { if (endpoints_prop[n.second].wsi == wsi) { atomic_store(&lws_service_cancelled, 1); @@ -169,20 +155,25 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s concurrent.erase(n.second); lwsl_err("CLIENT_CONNECTION_ERROR Unknown WIS: %s\n", in ? (char *)in : "(null)"); + pthread_mutex_unlock(&lock_concurrent); return -1; } } + pthread_mutex_unlock(&lock_concurrent); break; case LWS_CALLBACK_CLIENT_CLOSED: + pthread_mutex_lock(&lock_concurrent); lwsl_err("CLIENT_CALLBACK_CLIENT_CLOSED: %s\n", in ? (char *)in : ""); for (std::pair n : concurrent) { if (endpoints_prop[n.second].wsi == wsi) { m = n.second; + pthread_mutex_unlock(&lock_concurrent); goto do_retry; } } + pthread_mutex_unlock(&lock_concurrent); break; default : @@ -198,9 +189,11 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s &endpoints_prop[m].retry_count)) { if(endpoints_prop[m].retry_count > (LWS_ARRAY_SIZE(backoff_ms))){ + pthread_mutex_lock(&lock_concurrent); endpoints_prop.erase(m); concurrent.erase(m); atomic_store(&lws_service_cancelled, 1); + pthread_mutex_unlock(&lock_concurrent); return -1; } { @@ -311,7 +304,6 @@ void binance::Websocket::init() info.uid = -1; info.protocols = protocols; info.fd_limit_per_thread = 0; - info.extensions = extensions; context = lws_create_context(&info); if (!context) { From 55708075292071ba3367b1b53aa7c4e4a6acb465 Mon Sep 17 00:00:00 2001 From: MussoNero Date: Mon, 19 Oct 2020 00:02:49 +0200 Subject: [PATCH 7/9] Use wsi user_data instead of concurrency mapping --- include/binance_websocket.h | 2 +- src/binance_websocket.cpp | 199 ++++++++++++++++++------------------ 2 files changed, 102 insertions(+), 99 deletions(-) diff --git a/include/binance_websocket.h b/include/binance_websocket.h index 15b3ba8..bb72469 100644 --- a/include/binance_websocket.h +++ b/include/binance_websocket.h @@ -23,7 +23,7 @@ namespace binance public : static void connect_endpoint(CB user_cb, const char* path); static void init(); - static void enter_event_loop(std::chrono::hours hours = std::chrono::hours(24)); + static void enter_event_loop(const std::chrono::hours &hours = std::chrono::hours(24)); }; } diff --git a/src/binance_websocket.cpp b/src/binance_websocket.cpp index c72df76..db90b2c 100644 --- a/src/binance_websocket.cpp +++ b/src/binance_websocket.cpp @@ -12,14 +12,15 @@ #include #include #include +#include +#include using namespace binance; using namespace std; static struct lws_context *context; static atomic lws_service_cancelled(0); -void connect_client(lws_sorted_usec_list_t *sul); - +static void connect_client(lws_sorted_usec_list_t *sul); /* * This "contains" the endpoint connection proprty and has * the connection bound to it @@ -33,9 +34,8 @@ struct endpoint_connection { char* ws_path; }; -static std::map concurrent; -static std::map endpoints_prop; -static pthread_mutex_t lock_concurrent; /* serialize access */ +static std::map endpoints_prop; /* serialize access */ +static pthread_mutex_t lock_concurrent; /* lock serialize access */ /* * The retry and backoff policy we want to use for our client connections @@ -61,25 +61,26 @@ const lws_retry_bo_t retry = { */ }; -int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) +static int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { - int m; + std::atomic idx(-1); + auto *current_data = static_cast< endpoint_connection *>(user); switch (reason) { case LWS_CALLBACK_CLIENT_ESTABLISHED : - pthread_mutex_lock(&lock_concurrent); - for (std::pair n : concurrent) { - if (endpoints_prop[n.second].wsi == wsi) { + for (std::pair n : endpoints_prop) { + if (endpoints_prop[n.first].wsi == wsi && current_data->ws_path == endpoints_prop[n.first].ws_path) { + pthread_mutex_lock(&lock_concurrent); + idx = n.first; lws_callback_on_writable(wsi); - m = n.second; - lwsl_user("%s: connection established with success concurrent:%d ws_path::%s\n", - __func__, n.second, endpoints_prop[n.second].ws_path); + endpoints_prop[idx].wsi = wsi; + lwsl_user("%s: connection established with success endpoint#:%d ws_path::%s\n", + __func__, idx.load(), endpoints_prop[n.first].ws_path); pthread_mutex_unlock(&lock_concurrent); break; } } - pthread_mutex_unlock(&lock_concurrent); break; case LWS_CALLBACK_CLIENT_RECEIVE : @@ -87,24 +88,21 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s // Handle incomming messages here. try { - pthread_mutex_lock(&lock_concurrent); - string str_result = string(reinterpret_cast(in), len); - Json::Reader reader; - Json::Value json_result; - reader.parse(str_result , json_result); - - for (std::pair n : concurrent) { - if (endpoints_prop[n.second].wsi == wsi) { - m = n.second; - endpoints_prop[n.second].json_cb(json_result); - endpoints_prop[m].retry_count = 0; - lwsl_user("%s: incomming messages from %s\n", - __func__ , endpoints_prop[n.second].ws_path); - pthread_mutex_unlock(&lock_concurrent); + for (std::pair n : endpoints_prop) { + if (endpoints_prop[n.first].wsi == wsi && current_data->ws_path == endpoints_prop[n.first].ws_path) { + pthread_mutex_lock(&lock_concurrent); + string str_result = string(reinterpret_cast(in), len); + Json::Reader reader; + Json::Value json_result; + reader.parse(str_result , json_result); + assert(!json_result.isNull()); + idx = n.first; + endpoints_prop[idx].json_cb(json_result); + endpoints_prop[idx].retry_count = 0; + pthread_mutex_unlock(&lock_concurrent); break; } } - pthread_mutex_unlock(&lock_concurrent); } catch (exception &e) { @@ -119,17 +117,14 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s break; case LWS_CALLBACK_CLOSED : - pthread_mutex_lock(&lock_concurrent); - lwsl_err("CALLBACK_CLOSED: %s\n", - in ? (char *)in : ""); - for (std::pair n : concurrent) { - if (endpoints_prop[n.second].wsi == wsi) { - m = n.second; - pthread_mutex_unlock(&lock_concurrent); + for (std::pair n : endpoints_prop) { + if (endpoints_prop[n.first].wsi == wsi && current_data->ws_path == endpoints_prop[n.first].ws_path) { + idx = n.first; + lwsl_user("CALLBACK_CLOSED: %s\n", + in ? (char *)in : ""); goto do_retry; } } - pthread_mutex_unlock(&lock_concurrent); break; case LWS_CALLBACK_GET_THREAD_ID: @@ -147,33 +142,31 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s break; case LWS_CALLBACK_CLIENT_CONNECTION_ERROR : - pthread_mutex_lock(&lock_concurrent); - for (std::pair n : concurrent) { - if (endpoints_prop[n.second].wsi == wsi) { - atomic_store(&lws_service_cancelled, 1); - endpoints_prop.erase(n.second); - concurrent.erase(n.second); - lwsl_err("CLIENT_CONNECTION_ERROR Unknown WIS: %s\n", + for (std::pair n : endpoints_prop) { + if (endpoints_prop[n.first].wsi == wsi && current_data->ws_path == endpoints_prop[n.first].ws_path) { + pthread_mutex_lock(&lock_concurrent); + lwsl_err("CLIENT_CONNECTION_ERROR: %s\n", in ? (char *)in : "(null)"); - pthread_mutex_unlock(&lock_concurrent); + idx = n.first; + endpoints_prop.erase(idx); + lws_cancel_service(lws_get_context(wsi)); + lws_context_destroy(lws_get_context(wsi)); + atomic_store(&lws_service_cancelled, 1); + pthread_mutex_unlock(&lock_concurrent); return -1; } } - pthread_mutex_unlock(&lock_concurrent); break; case LWS_CALLBACK_CLIENT_CLOSED: - pthread_mutex_lock(&lock_concurrent); - lwsl_err("CLIENT_CALLBACK_CLIENT_CLOSED: %s\n", - in ? (char *)in : ""); - for (std::pair n : concurrent) { - if (endpoints_prop[n.second].wsi == wsi) { - m = n.second; - pthread_mutex_unlock(&lock_concurrent); + for (std::pair n : endpoints_prop) { + if (endpoints_prop[n.first].wsi == wsi && current_data->ws_path == endpoints_prop[n.first].ws_path) { + idx = n.first; + lwsl_user("CLIENT_CALLBACK_CLIENT_CLOSED: %s\n", + in ? (char *)in : ""); goto do_retry; } } - pthread_mutex_unlock(&lock_concurrent); break; default : @@ -185,27 +178,30 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s do_retry: try{ - if (lws_retry_sul_schedule_retry_wsi(endpoints_prop[m].wsi, &endpoints_prop[m]._sul, connect_client, - &endpoints_prop[m].retry_count)) + if (lws_retry_sul_schedule_retry_wsi(endpoints_prop[idx].wsi, &endpoints_prop[idx]._sul, connect_client, + &endpoints_prop[idx].retry_count)) { - if(endpoints_prop[m].retry_count > (LWS_ARRAY_SIZE(backoff_ms))){ + if(endpoints_prop[idx].retry_count > 2*(LWS_ARRAY_SIZE(backoff_ms))){ pthread_mutex_lock(&lock_concurrent); - endpoints_prop.erase(m); - concurrent.erase(m); + lwsl_err("%s: connection attempts exhausted, after [%d] retry, ws_path:%s\n", + __func__, endpoints_prop[idx].retry_count, endpoints_prop[idx].ws_path); + endpoints_prop.erase(idx); + lws_cancel_service(lws_get_context(wsi)); + lws_context_destroy(lws_get_context(wsi)); atomic_store(&lws_service_cancelled, 1); pthread_mutex_unlock(&lock_concurrent); return -1; } { - lwsl_err("%s: connection attempts exhausted,we will keep retrying count:%d ws_path:%s\n", - __func__, endpoints_prop[m].retry_count, endpoints_prop[m].ws_path); + lwsl_err("%s: connection attempts exhausted,we will keep retrying [%d] ws_path:%s\n", + __func__, endpoints_prop[idx].retry_count, endpoints_prop[idx].ws_path); atomic_store(&lws_service_cancelled, 0); - lws_sul_schedule(lws_get_context(endpoints_prop[m].wsi), 0, &endpoints_prop[m]._sul, connect_client, 10 * LWS_US_PER_SEC); + lws_sul_schedule(lws_get_context(endpoints_prop[idx].wsi), 0, &endpoints_prop[idx]._sul, connect_client, 1 * LWS_US_PER_MS); return 0; } } - lwsl_user("%s: connection attempts success, retrying count:%d ws_path:%s\n", - __func__, endpoints_prop[m].retry_count, endpoints_prop[m].ws_path); + lwsl_user("%s: connection attempts success, after [%d] retry, ws_path:%s\n", + __func__, endpoints_prop[idx].retry_count, endpoints_prop[idx].ws_path); }catch (exception &e) { Logger::write_log(" Error do_retry message : %s\n", e.what()); @@ -216,19 +212,19 @@ int event_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, s return 0; } -const lws_protocols protocols[] = +static const lws_protocols protocols[] = { { .name = "binance-websocket-api", .callback = event_cb, .per_session_data_size = 0, - .rx_buffer_size = 65536, + .rx_buffer_size = 0, }, { NULL, NULL, 0, 0 } /* end */ }; -void +static void sigint_handler(int sig) { Logger::write_log(" Interactive attention signal : %d\n", sig); @@ -238,12 +234,12 @@ sigint_handler(int sig) /* * Scheduled sul callback that starts the connection attempt */ -void connect_client(lws_sorted_usec_list_t *sul) +static void connect_client(lws_sorted_usec_list_t *sul) { - for (std::pair n : concurrent) { - if (&endpoints_prop[n.second]._sul == sul) { + for (std::pair n : endpoints_prop) { + if (&endpoints_prop[n.first]._sul == sul) { lwsl_user("%s: success ws_path::%s\n", - __func__, endpoints_prop[n.second].ws_path); + __func__, endpoints_prop[n.first].ws_path); struct lws_client_connect_info ccinfo; memset(&ccinfo, 0, sizeof(ccinfo)); @@ -251,35 +247,34 @@ void connect_client(lws_sorted_usec_list_t *sul) ccinfo.context = context; ccinfo.port = BINANCE_WS_PORT; ccinfo.address = BINANCE_WS_HOST; - ccinfo.path = endpoints_prop[n.second].ws_path; + ccinfo.path = endpoints_prop[n.first].ws_path; ccinfo.host = lws_canonical_hostname(context); ccinfo.origin = "origin"; ccinfo.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK | LCCSCF_PIPELINE; ccinfo.protocol = protocols[0].name; ccinfo.local_protocol_name = protocols[0].name; ccinfo.retry_and_idle_policy = &retry; - ccinfo.userdata = &endpoints_prop[n.second]; + ccinfo.userdata = &endpoints_prop[n.first]; /* * We store the new wsi here early in the connection process, * this gives the callback a way to identify which wsi faced the error * even before the new wsi is returned and even if ultimately no wsi is returned. */ - ccinfo.pwsi = &endpoints_prop[n.second].wsi; + ccinfo.pwsi = &endpoints_prop[n.first].wsi; - endpoints_prop[n.second].conn = lws_client_connect_via_info(&ccinfo); - if (!endpoints_prop[n.second].conn) + endpoints_prop[n.first].conn = lws_client_connect_via_info(&ccinfo); + if (!endpoints_prop[n.first].conn) { /* * Failed... schedule a retry... we can't use the _retry_wsi() * convenience wrapper api here because no valid wsi at this * point. */ - if (lws_retry_sul_schedule(context, 0, &endpoints_prop[n.second]._sul, &retry, - connect_client, &endpoints_prop[n.second].retry_count)) + if (lws_retry_sul_schedule(context, 0, &endpoints_prop[n.first]._sul, &retry, + connect_client, &endpoints_prop[n.first].retry_count)) { - lwsl_err("%s: connection attempts exhausted\n", __func__); - endpoints_prop.erase(n.second); - concurrent.erase(n.second); + lwsl_err("%s: Failed schedule a retry, we can't use the _retry_wsi():%s\n", + __func__, endpoints_prop[n.first].ws_path); atomic_store(&lws_service_cancelled, 1); return; } @@ -287,12 +282,13 @@ void connect_client(lws_sorted_usec_list_t *sul) break; } } - } void binance::Websocket::init() { pthread_mutex_init(&lock_concurrent, NULL); + endpoints_prop.clear(); + struct lws_context_creation_info info; signal(SIGINT, sigint_handler); memset(&info, 0, sizeof(info)); @@ -303,7 +299,8 @@ void binance::Websocket::init() info.gid = -1; info.uid = -1; info.protocols = protocols; - info.fd_limit_per_thread = 0; + info.fd_limit_per_thread = 1024; + info.max_http_header_pool = 1024; context = lws_create_context(&info); if (!context) { @@ -319,46 +316,52 @@ void binance::Websocket::init() void binance::Websocket::connect_endpoint(CB cb, const char* path) { pthread_mutex_lock(&lock_concurrent); - if(concurrent.size() > 1024){ + if(endpoints_prop.size() > 1024){ lwsl_err("%s: maximum of 1024 connect_endpoints reached,\n", - __func__); + __func__); pthread_mutex_unlock(&lock_concurrent); return; } - int n = concurrent.size(); - concurrent.emplace(std::pair(n,n)); + int n = endpoints_prop.size(); endpoints_prop[n].ws_path = const_cast(path); endpoints_prop[n].json_cb = cb; + pthread_mutex_unlock(&lock_concurrent); connect_client(&endpoints_prop[n]._sul); if (!lws_service_cancelled) { /* schedule the first client connection attempt to happen immediately */ - lws_sul_schedule(context, 0, &endpoints_prop[n]._sul, connect_client, 1); - lwsl_user("%s: concurrent:%d ws_path::%s\n", - __func__, n, endpoints_prop[n].ws_path); + lws_sul_schedule(context, 0, &endpoints_prop[n]._sul, connect_client, 1 * LWS_US_PER_MS); + lwsl_user("%s: schedule the first client connection for endpoint#:%d ws_path::%s\n", + __func__, n, endpoints_prop[n].ws_path); } - pthread_mutex_unlock(&lock_concurrent); } // Entering event loop -void binance::Websocket::enter_event_loop(std::chrono::hours hours) +void binance::Websocket::enter_event_loop(const std::chrono::hours &hours) { auto start = std::chrono::steady_clock::now(); auto end = start + hours; auto n = 0; do { - n = lws_service(context, 1000); - if (lws_service_cancelled) - { + try{ + n = lws_service(context, 500); + if (lws_service_cancelled) + { + lws_cancel_service(context); + break; + } + }catch ( exception &e ) { + lwsl_err("%s:::%s\n", + __func__, e.what()); + Logger::write_log( " Error ! %s", e.what() ); lws_cancel_service(context); break; } } while (n >= 0 && std::chrono::steady_clock::now() < end); - concurrent.clear(); + endpoints_prop.clear(); atomic_store(&lws_service_cancelled, 1); - lws_context_destroy(context); pthread_mutex_destroy(&lock_concurrent); -} +} \ No newline at end of file From ead9be8af41dd6b5eabd17eea7643968c237c365 Mon Sep 17 00:00:00 2001 From: mussonero Date: Wed, 22 Sep 2021 16:50:01 +0200 Subject: [PATCH 8/9] Update README.md --- README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/README.md b/README.md index 79b2f4b..d4db955 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,8 @@ + ``` +new version https://github.com/mussonero/binance-cxx-api-v2 + + ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿ ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⢿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿ ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⠟⠁⠀⠉⠻⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿ From fb1ce5cf6064ea6435ecee1668986f5c975856cb Mon Sep 17 00:00:00 2001 From: mussonero Date: Wed, 22 Sep 2021 16:51:03 +0200 Subject: [PATCH 9/9] Update README.md --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index d4db955..b8ad573 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,9 @@ -``` new version https://github.com/mussonero/binance-cxx-api-v2 +``` + ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿ ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⢿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿ ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⠟⠁⠀⠉⠻⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿