1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
#!/usr/bin/env python2
#
# A small program that transmits stdin to several TCP connections
#
# Known to work with TM 2. Be careful when using TM 1 or 4: modulator might
# lose frame phase synchronisation when frames are dropped. Setting
# PACKET_SIZE to 4*6144 should avoid this problem.
#
# 2012, mpb
from socket import *
import threading
import Queue
import sys
import os
import time
PACKET_SIZE=6144
QUEUE_PACKET_CAPACITY=100 # represents 2.4 seconds of ETI data
class Connection(object):
def __init__(self, conn, addr):
print("Got new connection from {0}".format(addr))
self.addr = addr
self.q = Queue.Queue(PACKET_SIZE*QUEUE_PACKET_CAPACITY)
self.ch = ConnectionHandler(conn, self.q)
self.ch.start()
def send(self, data):
if self.ch.running:
try:
self.q.put_nowait(data)
except Queue.Full:
pass
return True
else:
return False
def join(self):
self.ch.join()
def terminate(self):
self.ch.running = False
self.ch.close()
class ConnectionHandler(threading.Thread):
def __init__(self, sock, queue):
self.sock = sock
self.queue = queue
self.running = True
threading.Thread.__init__(self)
def run(self):
print("Server for {0} created".format(self.sock.getpeername()))
# Disable Nagle's algorithm, s.t. the TCP stack does not
# put together small send()s
self.sock.setsockopt(IPPROTO_TCP, TCP_NODELAY, 1)
while self.running:
data = self.queue.get()
try:
self.sock.sendall(data)
except:
self.running = False
self.sock.close()
def close(self):
self.sock.close()
import fcntl
class DataSender(threading.Thread):
def __init__(self, connections):
print("DS starting.")
if True:
# make stdin a non-blocking file
fd = sys.stdin.fileno()
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
self.running = True
self.connections = connections
threading.Thread.__init__(self)
def run(self):
while self.running:
try:
data = sys.stdin.read(PACKET_SIZE)
except KeyboardInterrupt:
break
except:
time.sleep(0.005)
continue
removeconns = []
for c in self.connections:
if not c.send(data):
removeconns.append(c)
#print("DS: Put {0} bytes into {1} connections".format(len(data), len(self.connections)))
for c in removeconns:
c.terminate()
self.connections.remove(c)
if len(removeconns) != 0:
print("DS: Removed {0} connections".format(len(removeconns)))
del removeconns
print("DS: Bye")
def listener(port):
print("eti_tcp.py [{0}] starting...".format(os.getpid()))
ADDR = ("", port)
serv = socket(AF_INET, SOCK_STREAM)
# Can reuse a socket that is in TIME_WAIT
serv.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
serv.bind((ADDR))
serv.listen(3)
connections = []
ds = DataSender(connections)
ds.start()
while True:
try:
print("Main: accepting on port {0}...".format(port))
sock, addr = serv.accept()
connections.append(Connection(sock, addr))
except KeyboardInterrupt:
print("Interrupted")
break
print("Terminating {0} connections".format(len(connections)))
for c in connections:
c.terminate()
print("Joining connections")
for c in connections:
c.join()
print("Send Ctrl-D to close stdin")
ds.running = False
ds.join()
def usage():
print("""Usage:
{0} port
port: specifies on which TCP port to listen for incoming connections""".format(sys.argv[0]))
if len(sys.argv) != 2:
usage()
sys.exit(1)
else:
port = 0
try:
port = int(sys.argv[1])
except:
usage()
sys.exit(1)
listener(port)
|
