Attempt to use UNIX sockets for better performance

3 files changed, 57 insertions, 14 deletions

diff --git a/src/gr-fadingui/python/netsink.py b/src/gr-fadingui/python/netsink.py
index 06e376c..8851fe9 100644
--- a/src/gr-fadingui/python/netsink.py
+++ b/src/gr-fadingui/python/netsink.py
@@ -3,12 +3,16 @@
 #
 # Copyright 2021 Sara Cinzia Halter, Naoki Pross.
 
+import os
 import socket
 from urllib.parse import urlparse
 
 import numpy as np
 from gnuradio import gr
 
+from fadingui.logger import get_logger
+log = get_logger("netsink")
+
 class netsink(gr.sync_block):
     """
     Sink that sends the data over the network using UDP.
@@ -26,7 +30,6 @@ class netsink(gr.sync_block):
         dt = to_numpy[dtype]
         if vlen > 1:
             dt = np.dtype(dt, (vlen,))
-        print(dt)
 
         gr.sync_block.__init__(self,
             name="Network Sink",
@@ -34,9 +37,28 @@ class netsink(gr.sync_block):
             out_sig=None)
 
         # Create a socket and parse remote machine url
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.url = urlparse(address)
-        self.srv = (self.url.hostname, self.url.port)
+        self.srv = None
+
+        if self.url.scheme == "udp":
+            log.debug(f"Creating UDP socket to {self.srv}")
+            self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self.srv = (self.url.hostname, self.url.port)
+            self.socket.connect(self.srv)
+
+        elif self.url.scheme == "file":
+            log.debug(f"Creating UNIX file socket to {self.url.path}")
+            self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+            self.srv = self.url.path
+            try:
+                self.socket.connect(self.srv)
+            except FileNotFoundError:
+                log.error("Cannot find socket file, is the server (GUI) running?")
+                raise
+
+        else:
+            raise NotImplemented
+
 
     def send(self, data):
         """
@@ -46,7 +68,11 @@ class netsink(gr.sync_block):
         @return Number of bytes that were actually sent
         """
         assert type(data) == bytes
-        return self.socket.sendto(data, self.srv)
+        try:
+            return self.socket.sendto(data, self.srv)
+        except socket.error as err:
+            log.warn(f"No data was sent: {err}")
+            return 0
 
     def encode(self, data):
         """
@@ -62,7 +88,7 @@ class netsink(gr.sync_block):
 
     def work(self, input_items, output_items):
         # send only every k-th sample
-        inp = input_items[0][::3]
+        inp = input_items[0][::2]
         inp_len = len(inp)
         blocksize = 1024
 
diff --git a/src/gui/gui.py b/src/gui/gui.py
index 9f63452..b298fd8 100755
--- a/src/gui/gui.py
+++ b/src/gui/gui.py
@@ -45,13 +45,13 @@ show_debug()
 time_plot         = net.network_plot(url="udp://localhost:31415", dtype=float, \
                         nsamples=500, tag="time_plot", label="Time plot")
 channel_plot      = net.network_constellation_plot(url="udp://localhost:31416", \
-                        nsamples=200, tag="channel_plot", label="Channel")
+                        nsamples=512, tag="channel_plot", label="Channel")
 synchronized_plot = net.network_constellation_plot(url="udp://localhost:31417", \
-                        nsamples=200, tag="synchronized_plot",  label="Synchronized")
+                        nsamples=512, tag="synchronized_plot",  label="Synchronized")
 equalized_plot    = net.network_constellation_plot(url="udp://localhost:31418", \
-                        nsamples=200, tag="equalized_plot",  label="Equalized")
+                        nsamples=512, tag="equalized_plot",  label="Equalized")
 locked_plot       = net.network_constellation_plot(url="udp://localhost:31419", \
-                        nsamples=200, tag="locked_plot",  label="Locked")
+                        nsamples=512, tag="locked_plot",  label="Locked")
 
 constellation_plots = [channel_plot, synchronized_plot, equalized_plot, locked_plot]
 network_plots = [time_plot] + constellation_plots
@@ -195,7 +195,7 @@ with window(label="RX DSP Flow Graph", width=800, height=400, pos=(0,25), tag="r
 # Network plots
 
 def make_constellation_plot_window(plot, label):
-    with window(label=label, no_collapse=True,
+    with window(label=label, no_collapse=True, no_close=True, \
             width=plot_window_sizes[plot][0], \
             height=plot_window_sizes[plot][1], \
             pos=plot_window_positions[plot], \
diff --git a/src/gui/net.py b/src/gui/net.py
index 121cc76..1ddb1d0 100644
--- a/src/gui/net.py
+++ b/src/gui/net.py
@@ -1,3 +1,4 @@
+import os
 import select
 import socket
 from urllib.parse import urlparse
@@ -13,18 +14,34 @@ class udpsource:
     Creates an UDP listening socket
     """
     def __init__(self, url, dtype, timeout=0.05):
-        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.url = urlparse(url)
         self.dtype = dtype
         self.timeout = timeout
 
+        if self.url.scheme == "udp":
+            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        elif self.url.scheme == "file":
+            try:
+                os.unlink(self.url.path)
+            except OSError:
+                if os.path.exists(self.url.path):
+                    raise
+
+            self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+        else:
+            raise NotImplemented
+
     def __del__(self):
         self.sock.close()
 
     def bind(self):
         self.sock.setblocking(False)
-        self.sock.bind((self.url.hostname, self.url.port))
-        # self.sock.listen()
+        if self.url.scheme == "udp":
+            self.sock.bind((self.url.hostname, self.url.port))
+        elif self.url.scheme == "file":
+            self.sock.bind(self.url.path)
+
+        # self.sock.listen(1)
 
     def read(self, nblocks):
         # TODO: run in a separate thread (it will be painful to implement)
@@ -33,7 +50,7 @@ class udpsource:
             return None
 
         # read from socket
-        blocksize = 1024 * 4
+        blocksize = 1024
         string = ready[0].recv(nblocks * blocksize).decode("ascii")
 
         # decode string, remove empty values