Improved the performance of the ACLK. (#8391) (#8401)
* Switched back to lws internal scheduler, small fragment size. * Fixing comment from review
Andrew Moss committed
Mar 14, 2020 at 18:40 UTC
82913d84783dfa25121a0af313a72576c34884ab
3 files changed
+19
-19
aclk/aclk_lws_wss_client.c
+10
-12
@@ -27,7 +27,7 @@ void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *re
27
for(w=0, wb=0, write_b = engine_instance->write_buffer_head; write_b != NULL; write_b = write_b->next)
28
{
29
w++;
30
- wb += write_b->data_size;
30
+ wb += write_b->data_size - write_b->written;
31
}
32
*write_len = w;
33
*write_len_bytes = wb;
@@ -399,7 +399,7 @@ static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reas
399
{
400
UNUSED(user);
401
struct lws_wss_packet_buffer *data;
402
- int retval = 0, rc;
402
+ int retval = 0;
403
404
// Callback servicing is forced when we are closed from above.
405
if (engine_instance->upstream_reconnect_request) {
@@ -419,21 +419,19 @@ static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reas
419
data = engine_instance->write_buffer_head;
420
if (likely(data)) {
421
size_t bytes_left = data->data_size - data->written;
422
- if ( bytes_left > 65536 )
423
- bytes_left = 65536;
424
- rc = lws_write(wsi, data->data + LWS_PRE + data->written, bytes_left, LWS_WRITE_BINARY);
425
- error("lws_write(req=%u,written=%u) %zu of %zu",bytes_left, rc, data->written,data->data_size,rc);
426
- data->written += bytes_left;
422
+ if ( bytes_left > FRAGMENT_SIZE)
423
+ bytes_left = FRAGMENT_SIZE;
424
+ int n = lws_write(wsi, data->data + LWS_PRE + data->written, bytes_left, LWS_WRITE_BINARY);
425
+ if (n>=0)
426
+ data->written += n;
427
+ //error("lws_write(req=%u,written=%u) %zu of %zu",bytes_left, rc, data->written,data->data_size,rc);
428
if (data->written == data->data_size)
429
{
430
lws_wss_packet_buffer_pop(&engine_instance->write_buffer_head);
431
lws_wss_packet_buffer_free(data);
432
}
433
if (engine_instance->write_buffer_head)
433
- {
434
- error("Req write");
434
lws_callback_on_writable(engine_instance->lws_wsi);
436
- }
435
}
436
aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);
437
return retval;
@@ -544,10 +542,10 @@ void aclk_lws_wss_service_loop()
542
{
543
if (engine_instance)
544
{
547
- if (engine_instance->lws_wsi) {
545
+ /*if (engine_instance->lws_wsi) {
546
lws_cancel_service(engine_instance->lws_context);
547
lws_callback_on_writable(engine_instance->lws_wsi);
550
- }
548
+ }*/
549
lws_service(engine_instance->lws_context, 0);
550
}
551
}
aclk/aclk_lws_wss_client.h
+1
-1
@@ -79,5 +79,5 @@ void aclk_lws_connection_data_received();
79
void aclk_lws_connection_closed();
80
void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len);
81
82
-
82
+#define FRAGMENT_SIZE 4096
83
#endif
aclk/agent_cloud_link.c
+8
-6
@@ -1413,8 +1413,8 @@ void *aclk_main(void *ptr)
1413
static int first_init = 0;
1414
size_t write_q, write_q_bytes, read_q;
1415
lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1416
- info("loop state first_init_%d connected=%d connecting=%d wq=%zu (%zu-bytes) rq=%zu",
1417
- first_init, aclk_connected, aclk_connecting, write_q, read_q);
1416
+ //info("loop state first_init_%d connected=%d connecting=%d wq=%zu (%zu-bytes) rq=%zu",
1417
+ // first_init, aclk_connected, aclk_connecting, write_q, write_q_bytes, read_q);
1418
if (unlikely(!aclk_connected)) {
1419
if (unlikely(!first_init)) {
1420
aclk_try_to_connect(aclk_hostname, aclk_port, port_num);
@@ -1441,10 +1441,12 @@ void *aclk_main(void *ptr)
1441
}
1442
1443
_link_event_loop();
1444
- //sleep_usec(USEC_PER_MS * 50);
1445
- static int stress_counter = 0;
1446
- if (stress_counter++ % 100 == 0 && write_q==0)
1447
- aclk_send_stress_test(2000000);
1444
+ /*static int stress_counter = 0;
1445
+ if (write_q_bytes==0 && stress_counter ++ >5)
1446
+ {
1447
+ aclk_send_stress_test(8000000);
1448
+ stress_counter = 0;
1449
+ }*/
1450
1451
// TODO: Move to on-connect
1452
if (unlikely(!aclk_subscribed)) {