Repository: brainflow-dev/brainflow (language: python)

Issue:
Major packet loss when multicasting at high sample rates
I'm getting around 70-90% packet loss when multicasting with Cyton Wifi on Windows 10.

It's completely fine when it doesn't multicast, hitting around 1,000 samples per second. With Synthetic it's working fine with and without multicasting.

You can add this code between `data = board.get_board_data()` and `board.stop_stream()` in the [Python Get Data from a Board
](https://brainflow.readthedocs.io/en/stable/Examples.html#python-get-data-from-a-board) sample to check packet loss:

`samples_per_second = data.shape[1]`
`total_samples = samples_per_second / 10`
`packet_loss = "{:.2f}%".format((1 - (samples_per_second / (BoardShim.get_sampling_rate(args.board_id) * 10))) * 100)`
`print(f"Packet loss: {packet_loss} ({total_samples}/sps)")`

Run that script with Cyton Wifi using multicast in shell:
`get_data_from_board.py" --board-id 5 --ip-address 192.168.4.1 --ip-port 70000  --streamer-params streaming_board://225.1.1.1:6677`

Here's the result from capturing packets on Wireshark for 10 seconds with Cyton Wifi on multicast:

![image](https://user-images.githubusercontent.com/3446032/118799297-e5f31e80-b8e1-11eb-930e-f10390fffd70.png)






Discussion / hints:
Welcome to BrainFlow project and thanks for your contribution! We will try to fix your issue ASAP. Make sure that you have read about issue format in the docs.
Hi,

Could you explain a little where exactly do you see lost packages? In the first process which reads data from a board or in the second one where you use streaming board?

Also,
```
packet_loss = "{:.2f}%".format((1 - (samples_per_second / (BoardShim.get_sampling_rate(args.board_id) * 10))) * 100)
```
this doesnt look like a valid way to estimate package loss, you need to analyze package_num row instead
I would recommend to check buffer overflow as described here https://ref.onixs.biz/lost-multicast-packets-troubleshooting.html 
And tune it as described in this article, also probably to avoid buffer overflow it makes sense to run second process(with streaming board) before the main one(with multicast streamer)

I will try to set higher sampling rate for syntehtic board and reproduce it as well
>  In the first process which reads data from a board or in the second one where you use streaming board?

It's in the first process which reads the data. Initially, I thought it was the second one using the streaming board, but realised it was indeed the first after looking into it deeper.

> this doesnt look like a valid way to estimate package loss, you need to analyze package_num row instead

Oh yes, sorry about that. I did export all those variations to csv files and those not hitting the intended sample rate were missing/skipping chunks of package number ranges. For example, with multicast on at 1,000 samples per second, consecutive rows did not have package_num from 0-255 but would skip a couple dozen roughly.

For example:
0,...
1,...
2,...
31,...
32,...
33,...
34,...
57,...
58,...
59,...
60,...
...

> I would recommend to check buffer overflow as described here https://ref.onixs.biz/lost-multicast-packets-troubleshooting.html
And tune it as described in this article, also probably to avoid buffer overflow it makes sense to run second process(with streaming board) before the main one(with multicast streamer)

Thank you for those pointers, I will try those out soon and report back.
"It's in the first process which reads the data. Initially, I thought it was the second one using the streaming board, but realised it was indeed the first after looking into it deeper"

If its in the 1st process it means that packages from device are lost and in this case I dont fully understand how multicast can affect it...

Also, to read data from cyton wifi we use tcp which ensure package delivery so I dont get how its possible
> If it's in the 1st process it means that packages from device are lost and in this case I dont fully understand how multicast can affect it...

> Also, to read data from cyton wifi we use tcp which ensure package delivery so I dont get how its possible

I had similar thoughts, which makes this so odd. I've attached the network traffic captured by Wireshark for both Cyton Wifi and Synthetic which shows gaps in time on packets received with Cyton but consistent with Synthetic in case this helps.

[cyton-wifi-multicast-10-seconds.txt](https://github.com/brainflow-dev/brainflow/files/6513384/cyton-wifi-multicast-10-seconds.txt)
[synthetic-multicast-10-seconds.txt](https://github.com/brainflow-dev/brainflow/files/6513385/synthetic-multicast-10-seconds.txt)


> I would recommend to check buffer overflow as described here https://ref.onixs.biz/lost-multicast-packets-troubleshooting.html

I've run diagnostics on both network adaptor and OS buffer and nothing looks odd to me from the results. I didn't diagnose on app-level as the article didn't have instructions for Windows, and didn't tune either as it also just had instructions for Linux. I'll try to find ways to do this and report back.

These are the results when multicasting for 10 seconds with Cyton Wifi (before and after):

**Network adapter buffer**

![image](https://user-images.githubusercontent.com/3446032/118929699-30c77180-b988-11eb-8797-112e324dab4e.png)

**OS system network kernel buffers**

![image](https://user-images.githubusercontent.com/3446032/118929912-7421e000-b988-11eb-9d74-18ca171f12c5.png)

> And tune it as described in this article, also probably to avoid buffer overflow it makes sense to run second process(with streaming board) before the main one(with multicast streamer)

I tried this and no luck either.

Did you get a chance to run the synthetic board with a higher sample rate?
Could you run your tests for cyton wifi wo multicast?
If you lose packets in the first process multicast can not affect it at all
Here's the result for cyton wifi wo multicast:

[cyton-wifi-no-multicast-10-seconds.txt](https://github.com/brainflow-dev/brainflow/files/6515179/cyton-wifi-no-multicast-10-seconds.txt)

Exporting that to csv file and looking at the results, it's hitting about 1,000 samples per second.

I re-ran cyton wifi with multicast and included both TCP and UDP log this time:

[cyton-wifi-multicast-10-seconds-incl-tcp.txt](https://github.com/brainflow-dev/brainflow/files/6515400/cyton-wifi-multicast-10-seconds-incl-tcp.txt)

Could you send stored data instead wireshark dumps?(use read write sample for reference https://brainflow.readthedocs.io/en/stable/Examples.html#python-read-write-file)
it will be easier to estimate sampling rate and package losses based on that 
Here it is with multicasting:
[cyton-wifi-multicast.txt](https://github.com/brainflow-dev/brainflow/files/6516274/cyton-wifi-multicast.txt)

Without:
[cyton-wifi.txt](https://github.com/brainflow-dev/brainflow/files/6516280/cyton-wifi.txt)


good,
here I dont see lost packages in the first file(no gaps in package num column) but I do see less datapoints for some reason. Its from the 1st process right?(not from streaming board)
Yes, from the first process.
I'll clarify just in case, it's the first process specified here:
https://brainflow.readthedocs.io/en/stable/SupportedBoards.html#streaming-board

```
params = BrainFlowInputParams()
params.ip_address = '192.168.4.1'
params.ip_port = 70000
board = BoardShim(BoardIds.CYTON_WIFI_BOARD.value, params)
board.prepare_session()
streaming_board = 'streaming_board://225.1.1.1:6677'
board.start_stream(450000, streaming_board)
```
Its weird since multicast streamer just calls `sendto` I didnt expect that this single call can be slow...
I can add it to another thread or add buffering for packages in multicast streamer and more likely it will solve the problem
One of those approaches should work from what I’m understanding too researching this.

Fix patch:
diff --git a/src/board_controller/board.cpp b/src/board_controller/board.cpp
index 090b96772..1168ebd7f 100644
--- a/src/board_controller/board.cpp
+++ b/src/board_controller/board.cpp
@@ -243,6 +243,8 @@ int Board::prepare_streamer (char *streamer_params)
                 safe_logger (spdlog::level::err, e.what ());
                 return (int)BrainFlowExitCodes::INVALID_ARGUMENTS_ERROR;
             }
+            safe_logger (spdlog::level::trace, "MultiCast Streamer, ip addr: {}, port: {}",
+                streamer_dest.c_str (), streamer_mods.c_str ());
             streamer = new MultiCastStreamer (streamer_dest.c_str (), port, num_rows);
         }
 
diff --git a/src/board_controller/inc/multicast_streamer.h b/src/board_controller/inc/multicast_streamer.h
index c8730edea..bffefea4c 100644
--- a/src/board_controller/inc/multicast_streamer.h
+++ b/src/board_controller/inc/multicast_streamer.h
@@ -1,7 +1,9 @@
 #pragma once
 
-#include "multicast_server.h"
+#include <thread>
 
+#include "data_buffer.h"
+#include "multicast_server.h"
 #include "streamer.h"
 
 
@@ -14,9 +16,18 @@ class MultiCastStreamer : public Streamer
 
     int init_streamer ();
     void stream_data (double *data);
+    static int get_packages_in_chunk ()
+    {
+        return 3;
+    }
 
 private:
     char ip[128];
     int port;
     MultiCastServer *server;
+    DataBuffer *db;
+    volatile bool is_streaming;
+    std::thread streaming_thread;
+
+    void thread_worker ();
 };
diff --git a/src/board_controller/multicast_streamer.cpp b/src/board_controller/multicast_streamer.cpp
index eeee24bc9..8e1a6b3b9 100644
--- a/src/board_controller/multicast_streamer.cpp
+++ b/src/board_controller/multicast_streamer.cpp
@@ -11,30 +11,84 @@ MultiCastStreamer::MultiCastStreamer (const char *ip, int port, int data_len) :
     strcpy (this->ip, ip);
     this->port = port;
     server = NULL;
+    is_streaming = false;
+    db = NULL;
 }
 
 MultiCastStreamer::~MultiCastStreamer ()
 {
+    if ((streaming_thread.joinable ()) && (is_streaming))
+    {
+        is_streaming = false;
+        streaming_thread.join ();
+    }
     if (server != NULL)
     {
         delete server;
         server = NULL;
     }
+    if (db != NULL)
+    {
+        delete db;
+        db = NULL;
+    }
 }
 
 int MultiCastStreamer::init_streamer ()
 {
+    if ((is_streaming) || (server != NULL) || (db != NULL))
+    {
+        Board::board_logger->error ("multicast streamer is running");
+        return (int)BrainFlowExitCodes::GENERAL_ERROR;
+    }
+
     server = new MultiCastServer (ip, port);
     int res = server->init ();
     if (res != (int)MultiCastReturnCodes::STATUS_OK)
     {
+        delete server;
+        server = NULL;
         Board::board_logger->error ("failed to init server multicast socket {}", res);
         return (int)BrainFlowExitCodes::GENERAL_ERROR;
     }
+
+    db = new DataBuffer (len, 1000);
+    if (!db->is_ready ())
+    {
+        Board::board_logger->error ("unable to prepare buffer for multicast");
+        delete db;
+        db = NULL;
+        delete server;
+        server = NULL;
+        return (int)BrainFlowExitCodes::INVALID_BUFFER_SIZE_ERROR;
+    }
+
+    is_streaming = true;
+    streaming_thread = std::thread ([this] { this->thread_worker (); });
     return (int)BrainFlowExitCodes::STATUS_OK;
 }
 
 void MultiCastStreamer::stream_data (double *data)
 {
-    server->send (data, sizeof (double) * len);
+    db->add_data (data);
+}
+
+void MultiCastStreamer::thread_worker ()
+{
+    int num_packages = MultiCastStreamer::get_packages_in_chunk ();
+    int transaction_len = num_packages * len;
+    double *transaction = new double[transaction_len];
+    for (int i = 0; i < transaction_len; i++)
+    {
+        transaction[i] = 0.0;
+    }
+    while (is_streaming)
+    {
+        if (db->get_data_count () >= (size_t)num_packages)
+        {
+            db->get_data (num_packages, transaction);
+            server->send (transaction, sizeof (double) * transaction_len);
+        }
+    }
+    delete[] transaction;
 }
diff --git a/src/board_controller/streaming_board.cpp b/src/board_controller/streaming_board.cpp
index 231c7cdde..4ce7a4531 100644
--- a/src/board_controller/streaming_board.cpp
+++ b/src/board_controller/streaming_board.cpp
@@ -1,6 +1,7 @@
 #include <string.h>
 
 #include "board_info_getter.h"
+#include "multicast_streamer.h"
 #include "streaming_board.h"
 
 #ifndef _WIN32
@@ -131,23 +132,28 @@ void StreamingBoard::read_thread ()
 {
     // format for incomming package is determined by original board
     int num_rows = board_descr["num_rows"];
-    int bytes_per_recv = sizeof (double) * num_rows;
-    double *package = new double[num_rows];
+    int num_packages = MultiCastStreamer::get_packages_in_chunk ();
+    int transaction_len = num_rows * num_packages;
+    int bytes_per_recv = sizeof (double) * transaction_len;
+    double *transaction = new double[transaction_len];
     for (int i = 0; i < num_rows; i++)
     {
-        package[i] = 0.0;
+        transaction[i] = 0.0;
     }
 
     while (keep_alive)
     {
-        int res = client->recv (package, bytes_per_recv);
+        int res = client->recv (transaction, bytes_per_recv);
         if (res != bytes_per_recv)
         {
             safe_logger (
                 spdlog::level::trace, "unable to read {} bytes, read {}", bytes_per_recv, res);
             continue;
         }
-        push_package (package);
+        for (int i = 0; i < num_packages; i++)
+        {
+            push_package (transaction + i * num_rows);
+        }
     }
-    delete[] package;
+    delete[] transaction;
 }
diff --git a/src/utils/multicast_client.cpp b/src/utils/multicast_client.cpp
index 6a294b812..8b25228db 100644
--- a/src/utils/multicast_client.cpp
+++ b/src/utils/multicast_client.cpp
@@ -41,9 +41,12 @@ int MultiCastClient::init ()
     // ensure that library will not hang in blocking recv/send call
     DWORD timeout = 5000;
     DWORD value = 1;
+    DWORD buf_size = 65000;
     setsockopt (client_socket, SOL_SOCKET, SO_REUSEADDR, (char *)&value, sizeof (value));
     setsockopt (client_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof (timeout));
     setsockopt (client_socket, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof (timeout));
+    setsockopt (client_socket, SOL_SOCKET, SO_SNDBUF, (const char *)&buf_size, sizeof (buf_size));
+    setsockopt (client_socket, SOL_SOCKET, SO_RCVBUF, (const char *)&buf_size, sizeof (buf_size));
 
     if (bind (client_socket, (const struct sockaddr *)&socket_addr, sizeof (socket_addr)) != 0)
     {
@@ -132,9 +135,12 @@ int MultiCastClient::init ()
     tv.tv_sec = 5;
     tv.tv_usec = 0;
     int value = 1;
+    int buf_size = 65000;
     setsockopt (client_socket, SOL_SOCKET, SO_REUSEADDR, &value, sizeof (value));
     setsockopt (client_socket, SOL_SOCKET, SO_RCVTIMEO, (const char *)&tv, sizeof (tv));
     setsockopt (client_socket, SOL_SOCKET, SO_SNDTIMEO, (const char *)&tv, sizeof (tv));
+    setsockopt (client_socket, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof (buf_size));
+    setsockopt (client_socket, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof (buf_size));
 
     if (bind (client_socket, (const struct sockaddr *)&socket_addr, sizeof (socket_addr)) != 0)
     {
diff --git a/src/utils/multicast_server.cpp b/src/utils/multicast_server.cpp
index aa4e2105a..07637d043 100644
--- a/src/utils/multicast_server.cpp
+++ b/src/utils/multicast_server.cpp
@@ -42,9 +42,11 @@ int MultiCastServer::init ()
     // ensure that library will not hang in blocking recv/send call
     DWORD timeout = 5000;
     DWORD value = 1;
+    DWORD buf_size = 65000;
     setsockopt (server_socket, SOL_SOCKET, SO_REUSEADDR, (char *)&value, sizeof (value));
     setsockopt (server_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof (timeout));
     setsockopt (server_socket, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof (timeout));
+    setsockopt (server_socket, SOL_SOCKET, SO_SNDBUF, (const char *)&buf_size, sizeof (buf_size));
 
     return (int)MultiCastReturnCodes::STATUS_OK;
 }
@@ -105,9 +107,12 @@ int MultiCastServer::init ()
     timeout.tv_sec = 5;
     timeout.tv_usec = 0;
     int value = 1;
+    int buf_size = 65000;
     setsockopt (server_socket, SOL_SOCKET, SO_REUSEADDR, (char *)&value, sizeof (value));
     setsockopt (server_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof (timeout));
     setsockopt (server_socket, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof (timeout));
+    setsockopt (server_socket, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof (buf_size));
+    setsockopt (server_socket, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof (buf_size));
 
     return (int)MultiCastReturnCodes::STATUS_OK;
 }


Test patch:
diff --git a/tests/cpp/signal_processing_demo/src/band_power.cpp b/tests/cpp/signal_processing_demo/src/band_power.cpp
index fe7198e6c..7fd41482e 100644
--- a/tests/cpp/signal_processing_demo/src/band_power.cpp
+++ b/tests/cpp/signal_processing_demo/src/band_power.cpp
@@ -30,9 +30,9 @@ int main (int argc, char *argv[])
         board->start_stream ();
 
 #ifdef _WIN32
-        Sleep (5000);
+        Sleep (10000);
 #else
-        sleep (5);
+        sleep (10);
 #endif
 
         board->stop_stream ();
diff --git a/tests/cpp/signal_processing_demo/src/band_power_all.cpp b/tests/cpp/signal_processing_demo/src/band_power_all.cpp
index f2e06c110..c0c7541f1 100644
--- a/tests/cpp/signal_processing_demo/src/band_power_all.cpp
+++ b/tests/cpp/signal_processing_demo/src/band_power_all.cpp
@@ -30,9 +30,9 @@ int main (int argc, char *argv[])
         board->start_stream ();
 
 #ifdef _WIN32
-        Sleep (5000);
+        Sleep (10000);
 #else
-        sleep (5);
+        sleep (10);
 #endif
 
         board->stop_stream ();