|
1 # coding=utf8 |
|
2 |
|
3 import socket, time, sys |
|
4 from d import D |
|
5 from parms import Parms |
|
6 |
|
7 class Meter(): |
|
8 |
|
9 def __init__(self, d): |
|
10 self.d = D("{}, troughput measuring daemon".format(d.debid)) |
|
11 |
|
12 def run(self): |
|
13 self.d.log("started", sev=3) |
|
14 ssc, sc = None, None |
|
15 ssc = self.bindwait('', Parms.baseport) |
|
16 try: |
|
17 while True: |
|
18 self.d.log("accepting...", sev=3) |
|
19 sc = ssc.accept()[0] |
|
20 self.d.log("accepted", sev=3) |
|
21 n1 = 0 |
|
22 n0 = len(sc.recv(16 * 1024)) |
|
23 while n0 > 0: |
|
24 n1 = n1 + n0 |
|
25 if self.d.ll(5): self.d.log("n0={}, n1={}".format(n0, n1)) |
|
26 if n1 >= 16 * 1024: |
|
27 if self.d.ll(5): self.d.log("{} received, sending acknoledgement".format(n1)) |
|
28 sc.send(bytes("=>{:08d}".format(n1), "utf8")) |
|
29 n1 = 0 |
|
30 n0 = len(sc.recv(16 * 1024)) |
|
31 sc.close() |
|
32 except KeyboardInterrupt: |
|
33 pass |
|
34 except Exception as e: |
|
35 self.d.abendMsg("measuring", e=e) |
|
36 self.d.log("closing ssc...", sev=4) |
|
37 if sc: sc.close() |
|
38 if ssc: ssc.close() |
|
39 |
|
40 def bindwait(self, host, port): |
|
41 ssc = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
|
42 while True: |
|
43 try: |
|
44 ssc.bind((host, port)) |
|
45 break |
|
46 except Exception as e: |
|
47 if e.strerror == "Address already in use": |
|
48 self.d.log("Address {}:{} already in use, waiting 10 secs...".format(host, port)) |
|
49 time.sleep(10) |
|
50 continue |
|
51 self.d.abend("bind", e) |
|
52 sys.exit(1) |
|
53 ssc.listen(1) |
|
54 self.d.log("bound to {}:{}".format(host, port), sev=2) |
|
55 return ssc |