Changeset 20 for trunk/bin

Timestamp:

08/18/15 22:09:24 (9 years ago)

Author:

tim

Message:

major code refactoring, better organizing location of library functions

File:

• trunk/bin/sampler (modified) (10 diffs)

trunk/bin/sampler

@@ -30 +30 @@
 
 from nanownlib import *
+from nanownlib.platform import *
+from nanownlib.tcpts import *
 import nanownlib.storage
 
@@ -36 +38 @@
 parser.add_argument('-c', dest='cases', type=str, default='{"short":10000,"long":13000}',
                     help='JSON representation of echo timing cases.')
-parser.add_argument('--no-tcpts', action='store_true', help='Disbale TCP timestamp profiling')
+parser.add_argument('--no-tcpts', action='store_true', help='Disable TCP timestamp profiling')
 parser.add_argument('--no-control', action='store_true', help='Do not collect separate control data.  Instead, synthesize it from test and train data.')
 parser.add_argument('session_name', default=None,
@@ -49 +51 @@
 
 
-session_name = options.session_name
-num_samples=options.sample_count
-hostname = options.host
-port = options.port
-protocol = 'http'
+num_samples = options.sample_count
+hostname    = options.host
+port        = options.port
+protocol    = 'http'
 
 cases = json.loads(options.cases)
+db_file = "%s.db" % options.session_name
+db = nanownlib.storage.db(db_file)
+
 
 def extractReportedRuntime(headers, body):
@@ -64 +68 @@
 def sendRequest(data=None):
     method = 'GET'
-    path = '/data/clamav-audit-results.txt?t=' + data
+    path = '/?t=' + data
     url = "%s://%s:%d%s" % (protocol,hostname,port,path)
     headers = {}
@@ -74 +78 @@
         try:
             session = requests.Session()
-            response = session.send(req)
-            #print(repr(response.raw._original_response.local_address))
+            response = session.send(req, allow_redirects=False)
+            #print("sendRequest:", repr(response.raw._original_response.local_address))
             reported = extractReportedRuntime(response.headers, response.text)
             retry = False
@@ -82 +86 @@
             time.sleep(1.0)
             sys.stderr.write("ERROR: retrying...\n")
-    
+
+        
     return {'userspace_rtt':response.elapsed.microseconds*1000,
             'reported':reported,
@@ -97 +102 @@
 
 setCPUAffinity()
-setTCPTimestamps()
-host_ip = socket.gethostbyname(hostname) #XXX: what about multiple A records?
+time.sleep(0.25) # Give our process a chance to migrate to a different CPU if necessary
+setPowersave(False) # XXX: test this to see if it helps
+setLowLatency(True) # XXX: test this to see if it helps
+tcpts_previous = setTCPTimestamps(True)
+
+
+#XXX: what about multiple A records?
+#     perform this during a script generation step, measuring lowest latency server
+#     and hard-coding the IP and host name separately.  However, including all
+#     server IPs in comments
+host_ip = socket.gethostbyname(hostname) 
+
 
 meta = {'tcpts_mean':None,'tcpts_stddev':None,'tcpts_slopes':None}
@@ -125 +140 @@
                      % (meta['tcpts_mean'], meta['tcpts_stddev'], 100*meta['tcpts_stddev']/meta['tcpts_mean']))
 
-
-sniffer_fp = tempfile.NamedTemporaryFile('w+t')
-db_file = "%s.db" % session_name
-
-sniffer = startSniffer(host_ip, port, sniffer_fp.name)
-db = nanownlib.storage.db(db_file)
 db.addMeta(meta)
-time.sleep(0.5) # ensure sniffer is fully ready and our process is migrated
-
-if options.no_control:
-    num_control = 0
-else:
-    num_control = int(num_samples*2/5)
-
-num_train = int((num_samples-num_control)/3)
-num_test = num_samples-num_train-num_control
-
-sample_types = [('train',num_train),
-                ('train_null',num_control),
-                ('test',num_test)]
-
-sid = 0
-report_interval = 20
-start = time.time()
-next_report = start+report_interval
-for st,count in sample_types:
-    if sniffer.poll() != None:
-        sys.stderr.write('ERROR: Sniffer quit, sender exiting...\n')
-        break
+
+
+def findNextSampleID(db):
+    cursor = db.conn.cursor()
+    cursor.execute("SELECT max(sample) FROM probes")
+    row = cursor.fetchone()
+    if row != None and row[0] != None:
+        return row[0]+1
+
+    return 0
+
+
+def collectSamples(db, sample_type, count, sniffer):
+    sniffer.start()
+
+    if not sniffer.is_running():
+        sys.stderr.write('ERROR: Sniffer did not start...\n')
+        return
         
+    sid = findNextSampleID(db)
     for k in range(0,count):
         sample_order = list(cases.items())
         random.shuffle(sample_order)
-        if st.endswith('null'):
+        if sample_type.endswith('null'):
             for i in range(1,len(sample_order)):
                 sample_order[i] = (sample_order[i][0],sample_order[0][1])
             random.shuffle(sample_order)
-            #print('after', sample_order)
             
         results = []
@@ -168 +173 @@
         for i in range(len(sample_order)):
             results.append(fetch({'sample':sid, 'test_case':sample_order[i][0],
-                                  'type':st, 'tc_order':i, 'time_of_day':now},
+                                  'type':sample_type, 'tc_order':i, 'time_of_day':now},
                                  sample_order[i][1]))
 
-        #print(results)
+        print(results)
         db.addProbes(results)
         db.conn.commit()
         sid += 1
 
-        if (time.time() > next_report):
-            #s = time.time()
-            reportProgress(db, sample_types, start)
-            #print("reportProgress time:", time.time()-s)
-            next_report += report_interval
-
-print("probes complete in %f" % (time.time()-start))
-time.sleep(2.0) # Give sniffer a chance to collect remaining packets
-stopSniffer(sniffer)
-
-start = time.time()
-associatePackets(sniffer_fp, db)
-sniffer_fp.close()
-end = time.time()
-print("associate time:", end-start)
-
+    time.sleep(2.0) # Give sniffer a chance to collect remaining packets
+    sniffer.stop()
+    #print(sniffer.openPacketLog().read())
+    start = time.time()
+    associatePackets(sniffer.openPacketLog(), db)
+    end = time.time()
+    print("associate time:", end-start)
+    
+
+
+if options.no_control:
+    num_control = 0
+else:
+    num_control = int(num_samples*2/5)
+
+num_train = int((num_samples-num_control)/3)
+num_test = num_samples-num_train-num_control
+
+sample_types = [('train',num_train),
+                ('train_null',num_control),
+                ('test',num_test)]
+
+sniffer = snifferProcess(host_ip, port)
+for st,count in sample_types:
+    collectSamples(db, st,count,sniffer)
+
+
+#start = time.time()
+#report_interval = 20
+#next_report = start+report_interval
+#        if (time.time() > next_report):
+#            reportProgress(db, sample_types, start)
+#            next_report += report_interval
+
+    
 if options.no_control:
     print("TODO: implement control synthesizing!")
@@ -199 +223 @@
 end = time.time()
 print("analyzed %d probes' packets in: %f" % (num_probes, end-start))
+
+
+setPowersave(True) # XXX: test this to see if it actually helps
+setLowLatency(False) # XXX: test this to see if it actually helps
+setTCPTimestamps(tcpts_previous)