''' LD_LIBRARY_PATH=/home/honglei/norm_test/ ''' # sender/Win10: #--id 34252 --send "C:\Users\Admin\AppData\Roaming\Wing Pro 9" --repeat 1 --addr 224.1.2.4/6003 --txaddr 10.65.39.191/8002 --cc rate 5000 --ack auto --grttprobing active --debug 4 # receiver/Debian10 #--id 172042016 --recv "~/recvFiles" --addr 224.1.2.4/6003 --interface ens33 --unicast_nack --debug 4 import sys import os if os.name =='nt': os.add_dll_directory(os.path.dirname(__file__)) import argparse import traceback import logging from typing import Optional import enum class CCEnum(enum.Enum): CC = "cc" CCE = "cce" CCL = "ccl" Fixed = "rate" class FlushEnum(enum.Enum): none='none' passive='passive' active='active' class GrttEnum(enum.Enum): none='none' passive='passive' active='active' class EnumAction(argparse.Action): """ Argparse action for handling Enums from https://stackoverflow.com/questions/43968006/support-for-enum-arguments-in-argparse """ def __init__(self, **kwargs): # Pop off the type value enum_type = kwargs.pop("type", None) # Ensure an Enum subclass is provided if enum_type is None: raise ValueError("type must be assigned an Enum when using EnumAction") if not issubclass(enum_type, enum.Enum): raise TypeError("type must be an Enum when using EnumAction") # Generate choices from the Enum #kwargs.setdefault("choices", tuple(e.value for e in enum_type)) #if e!=CCEnum.rate else "rate " super(EnumAction, self).__init__(**kwargs) self._enum = enum_type def __call__(self, parser, namespace, values:list|str, option_string=None): # Convert value back into an Enum if isinstance(values,list): if len(values) ==2: value = int(values[1]) elif len(values) ==1: value = self._enum( values[0].lower()) else: value = self._enum(values.lower() ) setattr(namespace, self.dest, value) ''' norm.cpp: fprintf(stderr, "Usage: normCast {send &| recv } [silent {on|off}]\n" " [repeat [updatesOnly]] [id ]\n" " [addr [/]][txaddr [/]][txport ]\n" " [interface ][reuse][loopback]\n" " [ack auto|[,,...]] [segment ]\n" " [block ] [parity ] [auto ]\n" " [cc|cce|ccl|rate ] [rxloss ]\n" " [txloss ] [flush {none|passive|active}]\n" " [grttprobing {none|passive|active}] [grtt ]\n" " [ptos ] [processor ] [saveaborts]\n" " [sentprocessor ]\n" " [purgeprocessor ] [buffer ]\n" " [txsockbuffer ] [rxsockbuffer ]\n" " [debug ] [trace] [log ]\n"); ''' # Instantiate the parser def get_arg_options(): parser = argparse.ArgumentParser(description='Optional app description') ## Required positional argument parser.add_argument('--id', type=int, required=True, help='id ') parser.add_argument('--send', '-S',nargs='+', help='send ') parser.add_argument('--recv', '-R', help='recv ') parser.add_argument('--repeat','-r', type=int, default=60, help='[repeat [updatesOnly]]') parser.add_argument('--interface', '-i', help='addr [/]') parser.add_argument('--addr', '-a', help='addr [/]') parser.add_argument('--txaddr', '-t', help='txaddr [/]') parser.add_argument('--txport', type=int, help='txport ') parser.add_argument('--ack', nargs='+', #nargs='?', help='ack auto|[,,...') #To create an option that needs no value, set the action [docs] of it to 'store_const', 'store_true' or 'store_false' parser.add_argument('--loopback', action='store_true',help='') parser.add_argument('--segment', type=int, help='segment ') parser.add_argument('--block', type=int, help='[block ] ') parser.add_argument('--parity', type=int, help='[parity ]') parser.add_argument('--auto', type=int, help='[auto ]') parser.add_argument('--grttprobing', type=GrttEnum, action=EnumAction, help='[grttprobing {none|passive|active}]') parser.add_argument('--grtt', type=int, help='[grtt ]') parser.add_argument('--ptos', type=int, help='[ptos ]') parser.add_argument('--flush', type=FlushEnum, action=EnumAction, help='[flush {none|passive|active}]') parser.add_argument('--silent', action='store_true',help='') parser.add_argument('--unicast_nack', action='store_true', help='') parser.add_argument('--txloss', type=float, help='[txloss ]') parser.add_argument('--rxloss', type=float, help='[rxloss ]') parser.add_argument('--buffer', type=int, help='[buffer ]') parser.add_argument('--txsockbuffer', type=int, help=' [txsockbuffer ]') parser.add_argument('--rxsockbuffer', type=int, help='[rxsockbuffer ]') parser.add_argument('--debug', type=int, help='[debug ]') parser.add_argument('--cc',type=CCEnum, nargs='+', action=EnumAction, help='[cc|cce|ccl|rate ] ' ) args = parser.parse_args() return args import pynorm def create_session(instance:pynorm.Instance, opts:argparse.Namespace): ''' use opts to create a NORM session ''' sessionAddr ='224.1.2.3' sessionPort = 6003 if opts.addr: args = opts.addr.split('/') if len(args) >0: sessionAddr = args[0] if len(args) ==2: sessionPort = int(args[1]) session = instance.createSession(sessionAddr, sessionPort, localId=opts.id) sessionTxAddr:str = None sessionTxPort = 8002 if opts.txaddr: args = opts.txaddr.split('/') if len(args) >0: sessionTxAddr = args[0] if len(args) ==2: sessionTxPort = int(args[1]) if sessionTxAddr: session.setTxPort(txPort=sessionTxPort, txBindAddr=sessionTxAddr) if opts.interface: session.setMulticastInterface(opts.interface) if opts.loopback: session.setLoopback(loopbackEnable=True) if opts.silent: session.setSilentReceiver(True) if opts.unicast_nack: session.setDefaultUnicastNack(enable=True) autoAck:bool = False ackingNodeList:list[int] =[] if opts.ack: if 'auto' == opts.ack[0]: autoAck = True session.setAutoAckingNodes(pynorm.TrackingStatus.RECEIVERS) else: ackingNodeList = [ int(i) for i in opts.ack] for ackingNondeID in ackingNodeList: session.addAckingNode(ackingNondeID) if opts.txloss and opts.txloss>0: session.setTxLoss(opts.txloss) if opts.rxloss and opts.rxloss>0: session.setRxLoss(opts.rxloss) # Congestion Control if opts.cc: if isinstance(opts.cc,int): session.setTxRate(opts.cc) session.setEcnSupport(ecnEnable=False) else: if opts.cc == CCEnum.CC: # default TCP-friendly congestion control session.setEcnSupport(ecnEnable=False) elif opts.cc == CCEnum.CCE: #"wireless-ready" ECN-only congestion control session.setEcnSupport(ecnEnable=True, ignoreLoss=True) elif opts.cc == CCEnum.CCL: # "loss tolerant", non-ECN congestion control session.setEcnSupport(ecnEnable=False, ignoreLoss=False, tolerateLoss=True) defaultBufferSpace = 64*1024*1024 defaultTxSocketBufferSize = 4*1024*1024 defaultRxSocketBufferSize = 6*1024*1024 bufferSpace = opts.buffer if opts.buffer else defaultBufferSpace if opts.send: if opts.ack: session.setFlowControl(flowControlFactor=0) #// ack-based flow control enabled on command-line, so disable timer-based flow control session.setBackoffFactor(0) #FEC session.startSender(sessionId=opts.id, bufferSpace=bufferSpace, segmentSize=opts.segment if opts.segment else 1400, blockSize=opts.block if opts.block else 64, numParity=opts.parity if opts.parity else 0, fecId=0) if opts.auto: session.setAutoParity(opts.auto) session.setTxSocketBuffer(opts.txsockbuffer if opts.txsockbuffer else defaultTxSocketBufferSize ) if opts.recv: session.startReceiver(bufferSpace=bufferSpace) session.setRxSocketBuffer(opts.rxsockbuffer if opts.rxsockbuffer else defaultRxSocketBufferSize ) if opts.grttprobing: session.setGrttProbingMode(pynorm.ProbingMode[ opts.grttprobing.value.upper() ] ) return session from pynorm import EventType from typing import Iterable def listFiles( dirFileList:list[str]) -> Iterable[str]: for dirFile in dirFileList: if os.path.isfile(dirFile): yield dirFile elif os.path.isdir(dirFile): for filename in os.scandir(dirFile): if filename.is_file(): yield filename.path else: raise FileExistsError(f"{dirFile} is not a valid file/dir") import logging import ipaddress class NormCaster(): def __init__(self,instance:pynorm.Instance, opts:argparse.Namespace ): self.opts:argparse.Namespace = opts self.recvDir:Optional[str] = os.path.expanduser(opts.recv) if opts.recv else None self.instance:pynorm.Instance = instance self.session:pynorm.Session = create_session(instance, opts) self.fileIterator:Iterable[str] = listFiles(opts.send) self.is_running:bool = True self.pendingSendFilePath:Optional[str] = None # def addOneFile(self,session): ''' ''' if self.pendingSendFilePath is None: try: self.pendingSendFilePath:str = self.fileIterator.__next__() except StopIteration: self.is_running = False return obj:Optional[pynorm.Object] = session.fileEnqueue(self.pendingSendFilePath, info= self.pendingSendFilePath.replace("\\","/").encode() ) if obj: if self.opts.ack: session.setWatermark(obj,True) logging.info(f"add file:{self.pendingSendFilePath} {obj._object}") self.pendingSendFilePath = None # succeed enqued else: logging.warning(f"fileEnqueue: {sendFilePath} failure!") def handle_norm_event(self, event:pynorm.Event): session:pynorm.Session = event.session evtType:pynorm.EventType = event.type logging.info(evtType ) if evtType in (EventType.TX_QUEUE_EMPTY, EventType.TX_QUEUE_VACANCY): self.addOneFile(session) elif evtType == EventType.GRTT_UPDATED: pass elif evtType == EventType.TX_WATERMARK_COMPLETED: if pynorm.AckingStatus.SUCCESS == session.getAckingStatus(): logging.warning( "normCast: NORM_TX_WATERMARK_COMPLETED, NORM_ACK_SUCCESS"); self.addOneFile(session) else: logging.warning( "normCast: NORM_TX_WATERMARK_COMPLETED, _NOT_ NORM_ACK_SUCCESS"); obj = event.object if obj is None: session.resetWatermark() elif evtType == EventType.TX_FLUSH_COMPLETED: pass elif evtType == EventType.TX_OBJECT_PURGED: obj = event.object if obj and obj.type== pynorm.ObjectType.FILE: logging.info(f"normCast: send file purged: {obj.info.decode()}") elif evtType == EventType.TX_OBJECT_SENT: obj = event.object if obj and obj.type== pynorm.ObjectType.FILE: logging.info(f"initial send complete: {obj.info.decode()}") self.addOneFile(session) elif evtType == EventType.ACKING_NODE_NEW: sender = event.sender logging.info(f"normCast: new acking node: {sender.id} IP address{sender.address}") elif evtType == EventType.REMOTE_SENDER_INACTIVE: pass elif evtType == EventType.RX_OBJECT_ABORTED: obj = event.object if obj and obj.type is not pynorm.ObjectType.FILE: logging.error("normCast: received invalid object type?!") return filePath = event.object.filename event.object.cancel() #remove temparary file if recv aborted. os.remove(filePath) logging.info("") elif evtType == EventType.RX_OBJECT_INFO: pass elif evtType == EventType.RX_OBJECT_COMPLETED: obj = event.object if obj and obj.type is not pynorm.ObjectType.FILE: logging.error("normCast: received invalid object type?!") return file_path = event.object.info.decode() fileName = os.path.split(file_path)[-1] path = os.path.join( self.recvDir, fileName ) oldPath = event.object.filename logging.debug (f"{oldPath=}") try: if os.path.isfile(path): os.remove(path) os.rename(src=oldPath, dst=path) except Exception as e: logging.error ( traceback.format_exc() ) def getAckNodesStatus(self,session): isSuccess, nodeID, nodeAckStatus = session.getNextAckingNode() while isSuccess: ip=ipaddress.IPv4Address(nodeID) if pynorm.AckingStatus.SUCCESS == nodeAckStatus: logging.info( f"normCast: node {nodeID} (IP address: {str(ip)}) acnkowledged.") else: logging.info( f"normCast: node {nodeID} (IP address: {str(ip)}) failed to acnkowledge.") isSuccess, nodeID, nodeAckStatus = session.getNextAckingNode() def run(self): while self.is_running: event: Optional[pynorm.Event,bool,None] = self.instance.getNextEvent(timeout=self.opts.repeat) while event: self.handle_norm_event(event) event = self.instance.getNextEvent(timeout=self.opts.repeat) if event is None and self.opts.send: self.addOneFile(self.session) # means timeout logging.warning("normCast exits!") def main(argv): opts:argparse.Namespace = get_arg_options() instance = pynorm.Instance() if opts.debug: instance.setDebugLevel(level=pynorm.DebugLevel(opts.debug) ) if opts.recv: recvPath:str = os.path.expanduser( opts.recv ) if not os.path.isdir(recvPath): #create is not exists! os.makedirs( recvPath ) instance.setCacheDirectory( recvPath ) normCast:NormCaster = NormCaster(instance, opts) if opts.send: normCast.addOneFile(normCast.session) normCast.run() if __name__ == '__main__': logging.getLogger().setLevel(logging.INFO) sys.exit(main(sys.argv))