Skip to content

Commit

Permalink
Centralising CONN_ constants, unifying 'config'
Browse files Browse the repository at this point in the history
  • Loading branch information
@irl
irl committed Oct 18, 2016
1 parent eaa5dd2 commit ba2acb7
Show file tree
Hide file tree
Showing 4 changed files with 38 additions and 41 deletions.
7 changes: 7 additions & 0 deletions pathspider/base.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import threading
import multiprocessing as mp
import queue
from enum import Enum

from ipaddress import ip_address

Expand Down Expand Up @@ -89,6 +90,12 @@ def empty(self):
while self.acquire(blocking=False):
pass

class Conn(Enum):
OK = 0
FAILED = 1
TIMEOUT = 2
SKIPPED = 3

QUEUE_SIZE = 1000
QUEUE_SLEEP = 0.5

Expand Down
19 changes: 8 additions & 11 deletions pathspider/plugins/dscp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from pathspider.base import SynchronizedSpider
from pathspider.base import PluggableSpider
from pathspider.base import Conn
from pathspider.base import NO_FLOW

from pathspider.observer import Observer
Expand All @@ -21,13 +22,9 @@

Connection = collections.namedtuple("Connection", ["client", "port", "state"])
SpiderRecord = collections.namedtuple("SpiderRecord", ["ip", "rport", "port",
"host", "dscp",
"host", "config",
"connstate"])

CONN_OK = 0
CONN_FAILED = 1
CONN_TIMEOUT = 2

## Chain functions

def dscp_setup(rec, ip):
Expand Down Expand Up @@ -91,11 +88,11 @@ def _connect(self, sock, job):
sock.settimeout(self.conn_timeout)
sock.connect((job[0], job[1]))

return Connection(sock, sock.getsockname()[1], CONN_OK)
return Connection(sock, sock.getsockname()[1], Conn.OK)
except TimeoutError:
return Connection(sock, sock.getsockname()[1], CONN_TIMEOUT)
return Connection(sock, sock.getsockname()[1], Conn.TIMEOUT)
except OSError:
return Connection(sock, sock.getsockname()[1], CONN_FAILED)
return Connection(sock, sock.getsockname()[1], Conn.FAILED)

def connect(self, job, pcs, config):
"""
Expand Down Expand Up @@ -123,7 +120,7 @@ def post_connect(self, job, conn, pcs, config):
Create the SpiderRecord
"""

if conn.state == CONN_OK:
if conn.state == Conn.OK:
rec = SpiderRecord(job[0], job[1], conn.port, job[2], config, True)
else:
rec = SpiderRecord(job[0], job[1], conn.port, job[2], config, False)
Expand Down Expand Up @@ -162,11 +159,11 @@ def merge(self, flow, res):
"sp": res.port,
"dp": res.rport,
"connstate": res.connstate,
"dscp": res.dscp,
"config": res.config,
"observed": False }
else:
flow['connstate'] = res.connstate
flow['dscp'] = res.dscp
flow['config'] = res.config
flow['observed'] = True

logger.debug("Result: " + str(flow))
Expand Down
20 changes: 8 additions & 12 deletions pathspider/plugins/ecn.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@

from pathspider.base import SynchronizedSpider
from pathspider.base import PluggableSpider
from pathspider.base import Conn
from pathspider.base import NO_FLOW

from pathspider.observer import Observer
from pathspider.observer import basic_flow
from pathspider.observer import basic_count
Expand All @@ -23,13 +23,9 @@

Connection = collections.namedtuple("Connection", ["client", "port", "state", "tstart"])
SpiderRecord = collections.namedtuple("SpiderRecord", ["ip", "rport", "port",
"rank", "host", "ecnstate",
"rank", "host", "config",
"connstate", "tstart", "tstop"])

CONN_OK = 0
CONN_FAILED = 1
CONN_TIMEOUT = 2

USER_AGENT = "pathspider"

## Chain functions
Expand Down Expand Up @@ -113,11 +109,11 @@ def connect(self, job, pcs, config):
sock.settimeout(self.conn_timeout)
sock.connect((job_ip, job_port))

return Connection(sock, sock.getsockname()[1], CONN_OK, tstart)
return Connection(sock, sock.getsockname()[1], Conn.OK, tstart)
except TimeoutError:
return Connection(sock, sock.getsockname()[1], CONN_TIMEOUT, tstart)
return Connection(sock, sock.getsockname()[1], Conn.TIMEOUT, tstart)
except OSError:
return Connection(sock, sock.getsockname()[1], CONN_FAILED, tstart)
return Connection(sock, sock.getsockname()[1], Conn.FAILED, tstart)

def post_connect(self, job, conn, pcs, config):
"""
Expand All @@ -128,7 +124,7 @@ def post_connect(self, job, conn, pcs, config):

tstop = str(datetime.utcnow())

if conn.state == CONN_OK:
if conn.state == Conn.OK:
rec = SpiderRecord(job_ip, job_port, conn.port, job_rank, job_host, config, True, conn.tstart, tstop)
else:
rec = SpiderRecord(job_ip, job_port, conn.port, job_rank, job_host, config, False, conn.tstart, tstop)
Expand Down Expand Up @@ -169,7 +165,7 @@ def combine_flows(self, flow):
other_flow = self.comparetab.pop(dip)

# first has always ecn off, while the second has ecn on
flows = (flow, other_flow) if other_flow['ecnstate'] else (other_flow, flow)
flows = (flow, other_flow) if other_flow['config'] else (other_flow, flow)
conditions = []

# discard non-observed flows and flows with no syn observed
Expand Down Expand Up @@ -240,7 +236,7 @@ def merge(self, flow, res):
flow['rank'] = res.rank
flow['host'] = res.host
flow['connstate'] = res.connstate
flow['ecnstate'] = res.ecnstate
flow['config'] = res.config
flow['tstart'] = res.tstart
flow['tstop'] = res.tstop

Expand Down
33 changes: 15 additions & 18 deletions pathspider/plugins/tfo.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,9 @@

from pathspider.base import DesynchronizedSpider
from pathspider.base import PluggableSpider
from pathspider.base import NO_RESULT, NO_FLOW
from pathspider.base import Conn
from pathspider.base import NO_RESULT
from pathspider.base import NO_FLOW

from pathspider.observer import Observer
from pathspider.observer import basic_flow
Expand All @@ -25,15 +27,10 @@
"c0t", "c1t"])
TFOSpiderRecord = collections.namedtuple("TFOSpiderRecord",
["ip", "rport", "port",
"host", "tfostate",
"host", "config",
"c0t", "c1t",
"connstate", "rank"])

CONN_OK = 0
CONN_FAILED = 1
CONN_TIMEOUT = 2
CONN_SKIPPED = 3

USER_AGENT = "pathspider"

## Chain functions
Expand Down Expand Up @@ -198,19 +195,19 @@ def connect(self, job, pcs, config):
c0t = timer() - tt

job.append(False)
return TFOConnection(sock, sock.getsockname()[1], CONN_OK, c0t, c1t)
return TFOConnection(sock, sock.getsockname()[1], Conn.OK, c0t, c1t)
except TimeoutError:
job.append(True)
return TFOConnection(sock, sock.getsockname()[1], CONN_TIMEOUT, c0t, c1t)
return TFOConnection(sock, sock.getsockname()[1], Conn.TIMEOUT, c0t, c1t)
except OSError:
job.append(True)
return TFOConnection(sock, sock.getsockname()[1], CONN_FAILED, c0t, c1t)
return TFOConnection(sock, sock.getsockname()[1], Conn.FAILED, c0t, c1t)

# with TFO
if config == 1:
# skip if config zero failed
if job[4]:
return TFOConnection(None, None, CONN_SKIPPED, 0, 0)
return TFOConnection(None, None, Conn.SKIPPED, 0, 0)

# make a message
message = bytes("GET / HTTP/1.1\r\nhost: "+str(job[2])+"\r\n\r\n", "utf-8")
Expand All @@ -233,11 +230,11 @@ def connect(self, job, pcs, config):
sock.sendto(message, socket.MSG_FASTOPEN, (job[0], job[1])) # pylint: disable=no-member
c1t = timer() - tt

return TFOConnection(sock, sock.getsockname()[1], CONN_OK, c0t, c1t)
return TFOConnection(sock, sock.getsockname()[1], Conn.OK, c0t, c1t)
except TimeoutError:
return TFOConnection(sock, sock.getsockname()[1], CONN_TIMEOUT, c0t, c1t)
return TFOConnection(sock, sock.getsockname()[1], Conn.TIMEOUT, c0t, c1t)
except OSError:
return TFOConnection(sock, sock.getsockname()[1], CONN_FAILED, c0t, c1t)
return TFOConnection(sock, sock.getsockname()[1], Conn.FAILED, c0t, c1t)

def post_connect(self, job, conn, pcs, config):
# try not shutting down
Expand All @@ -246,7 +243,7 @@ def post_connect(self, job, conn, pcs, config):
# except:
# pass

if conn.state == CONN_SKIPPED:
if conn.state == Conn.SKIPPED:
return NO_RESULT

try:
Expand All @@ -255,7 +252,7 @@ def post_connect(self, job, conn, pcs, config):
pass

return TFOSpiderRecord(job[0], job[1], conn.port, job[2], config,
conn.c0t, conn.c1t, conn.state == CONN_OK, job[3])
conn.c0t, conn.c1t, conn.state == Conn.OK, job[3])


def create_observer(self):
Expand All @@ -275,12 +272,12 @@ def create_observer(self):
def merge(self, flow, res):
logger = logging.getLogger('tfo')
if flow == NO_FLOW:
flow = {"dip": res.ip, "sp": res.port, "dp": res.rport, "connstate": res.connstate, "tfostate": res.tfostate, "observed": False }
flow = {"dip": res.ip, "sp": res.port, "dp": res.rport, "connstate": res.connstate, "config": res.config, "observed": False }
else:
flow['connstate'] = res.connstate
flow['host'] = res.host
flow['rank'] = res.rank
flow['tfostate'] = res.tfostate
flow['config'] = res.config
flow['observed'] = True
flow['cookie0_time'] = res.c0t
flow['cookie1_time'] = res.c1t
Expand Down

0 comments on commit ba2acb7

Please sign in to comment.