Skip to content

Commit

Permalink
Map and serialize every endpoint connection
Browse files Browse the repository at this point in the history 
Note: a single connection can listen to a maximum of 1024 streams
  • Loading branch information
@mussonero
mussonero committed Oct 16, 2020
1 parent 2851b5f commit 86f81fe
Showing 1 changed file with 151 additions and 88 deletions.
239 changes: 151 additions & 88 deletions src/binance_websocket.cpp
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
/*
Author: tensaix2j
Date : 2017/10/15
C++ library for Binance API.
*/

Expand All @@ -12,14 +12,11 @@
#include <libwebsockets.h>
#include <map>
#include <csignal>
#include <cassert>

using namespace binance;
using namespace std;

static struct lws_context *context;
static map<lws*, CB> handles;
static lws_sorted_usec_list_t _sul;
static atomic<int> lws_service_cancelled(0);
void connect_client(lws_sorted_usec_list_t *sul);

Expand All @@ -32,9 +29,14 @@ struct endpoint_connection {
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;
lws_sorted_usec_list_t _sul;
};

static std::map<int,int> concurrent;
static std::map<int, endpoint_connection> endpoints_prop;
static pthread_mutex_t lock_concurrent; /* serialize access */

/*
* The retry and backoff policy we want to use for our client connections
Expand Down Expand Up @@ -86,17 +88,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 :
case LWS_CALLBACK_CLIENT_ESTABLISHED :
lwsl_user("%s: established\n", __func__);
lws_callback_on_writable(wsi);
endpoint_prop->wsi = wsi;
for (std::pair<int, int> n : concurrent) {
if (endpoints_prop[n.second].wsi == 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);
break;
}
}
break;

case LWS_CALLBACK_CLIENT_RECEIVE :
case LWS_CALLBACK_CLIENT_RECEIVE :
{
// Handle incomming messages here.
try
Expand All @@ -106,24 +115,37 @@ 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<int, int> n : concurrent) {
if (endpoints_prop[n.second].wsi == wsi) {
m = n.second;
endpoints_prop[n.second].json_cb(json_result);
break;
}
}
}
catch (exception &e)
{
Logger::write_log("<binance::Websocket::event_cb> Error parsing incoming message : %s\n", e.what());
return 1;
Logger::write_log("<binance::Websocket::event_cb> 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<int, int> 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.
Expand All @@ -138,19 +160,33 @@ 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<int, int> n : concurrent) {
if (endpoints_prop[n.second].wsi == wsi) {
lwsl_err("CLIENT_CONNECTION_ERROR: %s\n",
in ? (char *)in : "(null)");
lwsl_err("%s: num:%d ws_path::%s\n",
__func__, n.first, endpoints_prop[n.second].ws_path);
m = n.second;
goto do_retry;
}
}
{
lwsl_err("CLIENT_CONNECTION_ERROR Unknown WIS: %s\n",
in ? (char *)in : "(null)");
lws_cancel_service(lws_get_context(wsi));
}
break;

case LWS_CALLBACK_CLIENT_CLOSED:
lwsl_err("CLIENT_CALLBACK_CLIENT_CLOSED: %s\n",
in ? (char *)in : "");
goto do_retry;
for (std::pair<int, int> n : concurrent) {
if (endpoints_prop[n.second].wsi == wsi) {
m = n.second;
goto do_retry;
}
}
break;

default :
// Make compiler happy regarding unhandled enums.
Expand All @@ -161,16 +197,29 @@ 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 > (2*LWS_ARRAY_SIZE(backoff_ms))){
if (endpoints_prop[m].wsi == wsi)
{
endpoints_prop.erase(m);
concurrent.erase(m);
lws_cancel_service(lws_get_context(wsi));
atomic_store(&lws_service_cancelled, 1);
return -1;
}
}else{
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 ws_path::%s\n",
__func__, endpoints_prop[m].ws_path);
endpoints_prop[m].retry_count = 0;
}catch (exception &e)
{
Logger::write_log("<binance::Websocket::event_cb> Error do_retry message : %s\n", e.what());
Expand Down Expand Up @@ -205,53 +254,53 @@ 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<int, int> 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, sul, &retry,
connect_client, &endpoints_prop[n.second].retry_count))
{
lwsl_err("%s: connection attempts exhausted\n", __func__);
atomic_store(&lws_service_cancelled, 1);
endpoints_prop.erase(n.second);
concurrent.erase(n.second);
return;
}
}
break;
}
}else{
handles[endpoint_prop->conn] = endpoint_prop->callback_jason_func;
}

}

void binance::Websocket::init()
Expand Down Expand Up @@ -282,13 +331,26 @@ 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<char *>(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<int,int>(n,n));
endpoints_prop[n].ws_path = const_cast<char *>(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
Expand All @@ -298,15 +360,16 @@ void binance::Websocket::enter_event_loop(std::chrono::hours hours)
auto end = start + hours;
auto n = 0;
do {
n = lws_service(context, 1000);
n = lws_service(context, 500);
if (lws_service_cancelled)
{
lws_cancel_service(context);
break;
}
} 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);
}

0 comments on commit 86f81fe

Please sign in to comment.