''' /* Copyright (C) 2019 Intel Corporation. All rights reserved. * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception */ ''' import select import socket import queue from time import sleep import struct import threading import time from ctypes import * import json import logging import os attr_type_list = [ "ATTR_TYPE_BYTE", # = ATTR_TYPE_INT8 "ATTR_TYPE_SHORT",# = ATTR_TYPE_INT16 "ATTR_TYPE_INT", # = ATTR_TYPE_INT32 "ATTR_TYPE_INT64", "ATTR_TYPE_UINT8", "ATTR_TYPE_UINT16", "ATTR_TYPE_UINT32", "ATTR_TYPE_UINT64", "ATTR_TYPE_FLOAT", "ATTR_TYPE_DOUBLE", "ATTR_NONE", "ATTR_NONE", "ATTR_TYPE_BOOLEAN", "ATTR_TYPE_STRING", "ATTR_TYPE_BYTEARRAY" ] Phase_Non_Start = 0 Phase_Leading = 1 Phase_Type = 2 Phase_Size = 3 Phase_Payload = 4 class imrt_link_message(object): def __init__(self): self.leading = bytes([0x12, 0x34]) self.phase = Phase_Non_Start self.size_in_phase = 0 self.message_type = bytes() self.message_size = bytes() self.payload = bytes() self.msg = bytes() def set_recv_phase(self, phase): self.phase = phase def on_imrt_link_byte_arrive(self, ch): self.msg += ch if self.phase == Phase_Non_Start: if ch == b'\x12': self.set_recv_phase(Phase_Leading) else: return -1 elif self.phase == Phase_Leading: if ch == b'\x34': self.set_recv_phase(Phase_Type) else: self.set_recv_phase(Phase_Non_Start) return -1 elif self.phase == Phase_Type: self.message_type += ch self.size_in_phase += 1 if self.size_in_phase == 2: (self.message_type, ) = struct.unpack('!H', self.message_type) self.size_in_phase = 0 self.set_recv_phase(Phase_Size) elif self.phase == Phase_Size: self.message_size += ch self.size_in_phase += 1 if self.size_in_phase == 4: (self.message_size, ) = struct.unpack('!I', self.message_size) self.size_in_phase = 0 self.set_recv_phase(Phase_Payload) if self.message_size == b'\x00': self.set_recv_phase(Phase_Non_Start) return 0 self.set_recv_phase(Phase_Payload) elif self.phase == Phase_Payload: self.payload += ch self.size_in_phase += 1 if self.size_in_phase == self.message_size: self.set_recv_phase(Phase_Non_Start) return 0 return 2 return 1 def read_file_to_buffer(file_name): file_object = open(file_name, 'rb') buffer = None if not os.path.exists(file_name): logging.error("file {} not found.".format(file_name)) return "file not found" try: buffer = file_object.read() finally: file_object.close() return buffer def decode_attr_container(msg): attr_dict = {} buf = msg[26 : ] (total_len, tag_len) = struct.unpack('@IH', buf[0 : 6]) tag_name = buf[6 : 6 + tag_len].decode() buf = buf[6 + tag_len : ] (attr_num, ) = struct.unpack('@H', buf[0 : 2]) buf = buf[2 : ] logging.info("parsed attr:") logging.info("total_len:{}, tag_len:{}, tag_name:{}, attr_num:{}" .format(str(total_len), str(tag_len), str(tag_name), str(attr_num))) for i in range(attr_num): (key_len, ) = struct.unpack('@H', buf[0 : 2]) key_name = buf[2 : 2 + key_len - 1].decode() buf = buf[2 + key_len : ] (type_index, ) = struct.unpack('@c', buf[0 : 1]) attr_type = attr_type_list[int(type_index[0])] buf = buf[1 : ] if attr_type == "ATTR_TYPE_BYTE": # = ATTR_TYPE_INT8 (attr_value, ) = struct.unpack('@c', buf[0 : 1]) buf = buf[1 : ] # continue elif attr_type == "ATTR_TYPE_SHORT": # = ATTR_TYPE_INT16 (attr_value, ) = struct.unpack('@h', buf[0 : 2]) buf = buf[2 : ] # continue elif attr_type == "ATTR_TYPE_INT": # = ATTR_TYPE_INT32 (attr_value, ) = struct.unpack('@i', buf[0 : 4]) buf = buf[4 : ] # continue elif attr_type == "ATTR_TYPE_INT64": (attr_value, ) = struct.unpack('@q', buf[0 : 8]) buf = buf[8 : ] # continue elif attr_type == "ATTR_TYPE_UINT8": (attr_value, ) = struct.unpack('@B', buf[0 : 1]) buf = buf[1 : ] # continue elif attr_type == "ATTR_TYPE_UINT16": (attr_value, ) = struct.unpack('@H', buf[0 : 2]) buf = buf[2 : ] # continue elif attr_type == "ATTR_TYPE_UINT32": (attr_value, ) = struct.unpack('@I', buf[0 : 4]) buf = buf[4 : ] # continue elif attr_type == "ATTR_TYPE_UINT64": (attr_value, ) = struct.unpack('@Q', buf[0 : 8]) buf = buf[8 : ] # continue elif attr_type == "ATTR_TYPE_FLOAT": (attr_value, ) = struct.unpack('@f', buf[0 : 4]) buf = buf[4 : ] # continue elif attr_type == "ATTR_TYPE_DOUBLE": (attr_value, ) = struct.unpack('@d', buf[0 : 8]) buf = buf[8 : ] # continue elif attr_type == "ATTR_TYPE_BOOLEAN": (attr_value, ) = struct.unpack('@?', buf[0 : 1]) buf = buf[1 : ] # continue elif attr_type == "ATTR_TYPE_STRING": (str_len, ) = struct.unpack('@H', buf[0 : 2]) attr_value = buf[2 : 2 + str_len - 1].decode() buf = buf[2 + str_len : ] # continue elif attr_type == "ATTR_TYPE_BYTEARRAY": (byte_len, ) = struct.unpack('@I', buf[0 : 4]) attr_value = buf[4 : 4 + byte_len] buf = buf[4 + byte_len : ] # continue attr_dict[key_name] = attr_value logging.info(str(attr_dict)) return attr_dict class Request(): mid = 0 url = "" action = 0 fmt = 0 payload = "" payload_len = 0 sender = 0 def __init__(self, url, action, fmt, payload, payload_len): self.url = url self.action = action self.fmt = fmt # if type(payload) == bytes: # self.payload = bytes(payload, encoding = "utf8") # else: self.payload_len = payload_len if self.payload_len > 0: self.payload = payload def pack_request(self): url_len = len(self.url) + 1 buffer_len = url_len + self.payload_len req_buffer = struct.pack('!2BH2IHI',1, self.action, self.fmt, self.mid, self.sender, url_len, self.payload_len) for i in range(url_len - 1): req_buffer += struct.pack('!c', bytes(self.url[i], encoding = "utf8")) req_buffer += bytes([0]) for i in range(self.payload_len): req_buffer += struct.pack('!B', self.payload[i]) return req_buffer, len(req_buffer) def send(self, conn, is_install): leading = struct.pack('!2B', 0x12, 0x34) if not is_install: msg_type = struct.pack('!H', 0x0002) else: msg_type = struct.pack('!H', 0x0004) buff, buff_len = self.pack_request() lenth = struct.pack('!I', buff_len) try: conn.send(leading) conn.send(msg_type) conn.send(lenth) conn.send(buff) except socket.error as e: logging.error("device closed") for dev in tcpserver.devices: if dev.conn == conn: tcpserver.devices.remove(dev) return -1 def query(conn): req = Request("/applet", 1, 0, "", 0) if req.send(conn, False) == -1: return "fail" time.sleep(0.05) try: receive_context = imrt_link_message() start = time.time() while True: if receive_context.on_imrt_link_byte_arrive(conn.recv(1)) == 0: break elif time.time() - start >= 5.0: return "fail" query_resp = receive_context.msg print(query_resp) except OSError as e: logging.error("OSError exception occur") return "fail" res = decode_attr_container(query_resp) logging.info('Query device infomation success') return res def install(conn, app_name, wasm_file): wasm = read_file_to_buffer(wasm_file) if wasm == "file not found": return "failed to install: file not found" print("wasm file len:") print(len(wasm)) req = Request("/applet?name=" + app_name, 3, 98, wasm, len(wasm)) if req.send(conn, True) == -1: return "fail" time.sleep(0.05) try: receive_context = imrt_link_message() start = time.time() while True: if receive_context.on_imrt_link_byte_arrive(conn.recv(1)) == 0: break elif time.time() - start >= 5.0: return "fail" msg = receive_context.msg except OSError as e: logging.error("OSError exception occur") # TODO: check return message if len(msg) == 24 and msg[8 + 1] == 65: logging.info('Install application success') return "success" else: res = decode_attr_container(msg) logging.warning('Install application failed: %s' % (str(res))) print(str(res)) return str(res) def uninstall(conn, app_name): req = Request("/applet?name=" + app_name, 4, 99, "", 0) if req.send(conn, False) == -1: return "fail" time.sleep(0.05) try: receive_context = imrt_link_message() start = time.time() while True: if receive_context.on_imrt_link_byte_arrive(conn.recv(1)) == 0: break elif time.time() - start >= 5.0: return "fail" msg = receive_context.msg except OSError as e: logging.error("OSError exception occur") # TODO: check return message if len(msg) == 24 and msg[8 + 1] == 66: logging.info('Uninstall application success') return "success" else: res = decode_attr_container(msg) logging.warning('Uninstall application failed: %s' % (str(res))) print(str(res)) return str(res) class Device: def __init__(self, conn, addr, port): self.conn = conn self.addr = addr self.port = port self.app_num = 0 self.apps = [] cmd = [] class TCPServer: def __init__(self, server, server_address, inputs, outputs, message_queues): # Create a TCP/IP self.server = server self.server.setblocking(False) # Bind the socket to the port self.server_address = server_address print('starting up on %s port %s' % self.server_address) self.server.bind(self.server_address) # Listen for incoming connections self.server.listen(10) self.cmd_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.cmd_sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) self.cmd_sock.bind(('127.0.0.1', 8889)) self.cmd_sock.listen(5) # Sockets from which we expect to read self.inputs = inputs self.inputs.append(self.cmd_sock) # Sockets to which we expect to write # 处理要发送的消息 self.outputs = outputs # Outgoing message queues (socket: Queue) self.message_queues = message_queues self.devices = [] self.conn_dict = {} def handler_recever(self, readable): # Handle inputs for s in readable: if s is self.server: # A "readable" socket is ready to accept a connection connection, client_address = s.accept() self.client_address = client_address print('connection from', client_address) # this is connection not server # connection.setblocking(0) self.inputs.append(connection) # Give the connection a queue for data we want to send # self.message_queues[connection] = queue.Queue() res = query(connection) if res != "fail": dev = Device(connection, client_address[0], client_address[1]) self.devices.append(dev) self.conn_dict[client_address] = connection dev_info = {} dev_info['addr'] = dev.addr dev_info['port'] = dev.port dev_info['apps'] = 0 logging.info('A new client connected from ("%s":"%s")' % (dev.conn, dev.port)) elif s is self.cmd_sock: connection, client_address = s.accept() print("web server socket connected") logging.info("Django server connected") self.inputs.append(connection) self.message_queues[connection] = queue.Queue() else: data = s.recv(1024) if data != b'': # A readable client socket has data logging.info('received "%s" from %s' % (data, s.getpeername())) # self.message_queues[s].put(data) # # Add output channel for response # if s not in self.outputs: # self.outputs.append(s) if(data.decode().split(':')[0] == "query"): if data.decode().split(':')[1] == "all": resp = [] print('start query all devices') for dev in self.devices: dev_info = query(dev.conn) if dev_info == "fail": continue dev_info["addr"] = dev.addr dev_info["port"] = dev.port resp.append(str(dev_info)) print(resp) if self.message_queues[s] is not None: # '*' is used in web server to sperate the string self.message_queues[s].put(bytes("*".join(resp), encoding = 'utf8')) if s not in self.outputs: self.outputs.append(s) else: client_addr = (data.decode().split(':')[1],int(data.decode().split(':')[2])) if client_addr in self.conn_dict.keys(): print('start query device from (%s:%s)' % (client_addr[0], client_addr[1])) resp = query(self.conn_dict[client_addr]) print(resp) if self.message_queues[s] is not None: self.message_queues[s].put(bytes(str(resp), encoding = 'utf8')) if s not in self.outputs: self.outputs.append(s) else: # no connection if self.message_queues[s] is not None: self.message_queues[s].put(bytes(str("fail"), encoding = 'utf8')) if s not in self.outputs: self.outputs.append(s) elif(data.decode().split(':')[0] == "install"): client_addr = (data.decode().split(':')[1],int(data.decode().split(':')[2])) app_name = data.decode().split(':')[3] app_file = data.decode().split(':')[4] if client_addr in self.conn_dict.keys(): print('start install application %s to ("%s":"%s")' % (app_name, client_addr[0], client_addr[1])) res = install(self.conn_dict[client_addr], app_name, app_file) if self.message_queues[s] is not None: logging.info("response {} to cmd server".format(res)) self.message_queues[s].put(bytes(res, encoding = 'utf8')) if s not in self.outputs: self.outputs.append(s) elif(data.decode().split(':')[0] == "uninstall"): client_addr = (data.decode().split(':')[1],int(data.decode().split(':')[2])) app_name = data.decode().split(':')[3] if client_addr in self.conn_dict.keys(): print("start uninstall") res = uninstall(self.conn_dict[client_addr], app_name) if self.message_queues[s] is not None: logging.info("response {} to cmd server".format(res)) self.message_queues[s].put(bytes(res, encoding = 'utf8')) if s not in self.outputs: self.outputs.append(s) # if self.message_queues[s] is not None: # self.message_queues[s].put(data) # if s not in self.outputs: # self.outputs.append(s) else: logging.warning(data) # Interpret empty result as closed connection try: for dev in self.devices: if s == dev.conn: self.devices.remove(dev) # Stop listening for input on the connection if s in self.outputs: self.outputs.remove(s) self.inputs.remove(s) # Remove message queue if s in self.message_queues.keys(): del self.message_queues[s] s.close() except OSError as e: logging.error("OSError raised, unknown connection") return "got it" def handler_send(self, writable): # Handle outputs for s in writable: try: message_queue = self.message_queues.get(s) send_data = '' if message_queue is not None: send_data = message_queue.get_nowait() except queue.Empty: self.outputs.remove(s) else: # print "sending %s to %s " % (send_data, s.getpeername) # print "send something" if message_queue is not None: s.send(send_data) else: print("client has closed") # del message_queues[s] # writable.remove(s) # print "Client %s disconnected" % (client_address) return "got it" def handler_exception(self, exceptional): # # Handle "exceptional conditions" for s in exceptional: print('exception condition on', s.getpeername()) # Stop listening for input on the connection self.inputs.remove(s) if s in self.outputs: self.outputs.remove(s) s.close() # Remove message queue del self.message_queues[s] return "got it" def event_loop(tcpserver, inputs, outputs): while inputs: # Wait for at least one of the sockets to be ready for processing print('waiting for the next event') readable, writable, exceptional = select.select(inputs, outputs, inputs) if readable is not None: tcp_recever = tcpserver.handler_recever(readable) if tcp_recever == 'got it': print("server have received") if writable is not None: tcp_send = tcpserver.handler_send(writable) if tcp_send == 'got it': print("server have send") if exceptional is not None: tcp_exception = tcpserver.handler_exception(exceptional) if tcp_exception == 'got it': print("server have exception") sleep(0.1) def run_wasm_server(): server_address = ('localhost', 8888) server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) inputs = [server] outputs = [] message_queues = {} tcpserver = TCPServer(server, server_address, inputs, outputs, message_queues) task = threading.Thread(target=event_loop,args=(tcpserver,inputs,outputs)) task.start() if __name__ == '__main__': logging.basicConfig(level=logging.DEBUG, filename='wasm_server.log', filemode='a', format= '%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s' ) server_address = ('0.0.0.0', 8888) server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) inputs = [server] outputs = [] message_queues = {} tcpserver = TCPServer(server, server_address, inputs, outputs, message_queues) logging.info("TCP Server start at {}:{}".format(server_address[0], "8888")) task = threading.Thread(target=event_loop,args=(tcpserver,inputs,outputs)) task.start() # event_loop(tcpserver, inputs, outputs)