diff --git a/MAVProxy/modules/mavproxy_ftp.py b/MAVProxy/modules/mavproxy_ftp.py index 8ec1a371ac..6f90f92523 100644 --- a/MAVProxy/modules/mavproxy_ftp.py +++ b/MAVProxy/modules/mavproxy_ftp.py @@ -1,13 +1,15 @@ #!/usr/bin/env python3 '''mavlink file transfer support''' -import io -import time, os, sys import glob +import os +import sys +import time import struct import random import zlib -from pymavlink import mavutil +import heapq +import socket try: # py2 @@ -36,6 +38,9 @@ OP_Rename = 13 OP_CalcFileCRC32 = 14 OP_BurstReadFile = 15 +# Like OP_ListDirectory, with an mtime field appended to file entries. Older +# servers either NACK this request or ignore it, so callers must fall back. +OP_ListDirectoryWithTime = 16 OP_Ack = 128 OP_Nack = 129 @@ -54,6 +59,19 @@ HDR_Len = 12 MAX_Payload = 239 +# Four bytes make accidental capability detection from a legacy server's +# CreateFile reply vanishingly unlikely. The bytes on the wire spell MFQ1. +WRITE_CAPABILITY_MAGIC = b'MFQ1' + +# Keep network writes below a normal Ethernet MTU. MAVLink parsers accept +# multiple frames in one datagram, while avoiding IP fragmentation makes the +# whole batch much less likely to be lost on a poor link. +MAX_NETWORK_BATCH = 1200 + +# A retired session may still have packets in a real link's buffers. This is +# longer than ArduPilot's FTP session expiry and prevents a stale reply or +# request being routed to a new operation after the uint8 session ID wraps. +SESSION_REUSE_DELAY = 30.0 # the server null terminates the last byte of its name buffer, so a name # filling the payload exactly would be silently truncated @@ -100,29 +118,29 @@ def __init__(self, ofs, size): self.size = size self.last_send = 0 -class FTPModule(mp_module.MPModule): - def __init__(self, mpstate): - super(FTPModule, self).__init__(mpstate, "ftp", public=True) - self.add_command('ftp', self.cmd_ftp, "file transfer", - ["", - "set (FTPSETTING)", - "put (FILENAME) (FILENAME)", - "crclocal (FILENAME)", - "crccmp (FILENAME)"]) - self.ftp_settings = mp_settings.MPSettings( - [('debug', int, 0), - ('pkt_loss_tx', int, 0), - ('pkt_loss_rx', int, 0), - ('max_backlog', int, 5), - ('burst_read_size', int, 80), - ('write_size', int, 80), - ('write_qsize', int, 5), - ('retry_time', float, 0.5), - ('crccmp_timeout', float, 120.0)]) - self.add_completion_function('(FTPSETTING)', - self.ftp_settings.completion) + +class MAVLinkBatchWriter: + '''Collect MAVLink packets so several FTP requests use one link write.''' + def __init__(self): + self.packets = [] + + def write(self, packet): + self.packets.append(bytes(packet)) + return len(packet) + + +class FTPWorker(mp_module.MPModule): + '''State for one FTP operation/session. + + A worker deliberately isn't registered as a public MAVProxy module. The + public FTPModule owns and routes to several of these at once. + ''' + def __init__(self, manager, session): + super(FTPWorker, self).__init__(manager.mpstate, "ftp_worker") + self.manager = manager + self.ftp_settings = manager.ftp_settings self.seq = 0 - self.session = 0 + self.session = session self.network = 0 self.last_op = None self.fh = None @@ -142,8 +160,13 @@ def __init__(self, mpstate): self.last_burst_read = None self.op_start = None self.dir_offset = 0 + self.list_dname = None + self.list_with_time = False self.last_op_time = time.time() self.rtt = 0.5 + self.rttvar = 0.25 + self.rtt_valid = False + self.send_times = {} self.reached_eof = False self.backlog = 0 self.burst_size = self.ftp_settings.burst_read_size @@ -156,7 +179,13 @@ def __init__(self, mpstate): self.write_idx = 0 self.write_recv_idx = -1 self.write_pending = 0 + self.write_inflight = set() self.write_last_send = None + self.write_open = False + self.write_qsize = (self.ftp_settings.write_qsize + if self.ftp_settings.write_qsize > 0 else 5) + self.write_batch_size = (self.ftp_settings.write_batch_size + if self.ftp_settings.write_batch_size > 0 else 1) self.warned_component = False # console progress is only for interactive ftp get/put, not for the # callback-driven transfers behind "param ftp" and "wp ftp" @@ -174,70 +203,79 @@ def __init__(self, mpstate): self.crccmp_sent = None self.crccmp_start = 0 self.crccmp_expect = None - - def cmd_ftp(self, args): - '''FTP operations''' - usage = "Usage: ftp " - if len(args) < 1: - print(usage) - return - # these all talk to the vehicle and would have their replies consumed - # by a running comparison, or clobber the op it is waiting on - if self.crccmp_dest is not None and args[0] in ( - 'list', 'crc', 'rm', 'rmdir', 'rename', 'mkdir', 'put'): - print("crccmp in progress, use 'ftp cancel' to stop it") - return - if args[0] == 'list': - self.cmd_list(args[1:]) - elif args[0] == "set": - self.ftp_settings.command(args[1:]) - elif args[0] == 'get': - self.cmd_get(args[1:]) - elif args[0] == 'put': - self.cmd_put(args[1:]) - elif args[0] == 'rm': - self.cmd_rm(args[1:]) - elif args[0] == 'rmdir': - self.cmd_rmdir(args[1:]) - elif args[0] == 'rename': - self.cmd_rename(args[1:]) - elif args[0] == 'mkdir': - self.cmd_mkdir(args[1:]) - elif args[0] == 'crc': - self.cmd_crc(args[1:]) - elif args[0] == 'crclocal': - self.cmd_crclocal(args[1:]) - elif args[0] == 'crccmp': - self.cmd_crccmp(args[1:]) - elif args[0] == 'status': - self.cmd_status() - elif args[0] == 'cancel': - self.cmd_cancel() - else: - print(usage) - - def send(self, op): - '''send a request''' - op.seq = self.seq + self.session_waiting = False + self.session_wait_reported = False + self.done = False + self.last_op_reply = False + self.request_retries = 0 + + def prepare_send(self, op, preserve_seq=False): + '''prepare a request and update protocol state as if it was sent''' + if not preserve_seq: + op.seq = self.seq payload = op.pack() plen = len(payload) if plen < MAX_Payload + HDR_Len: payload.extend(bytearray([0]*((HDR_Len+MAX_Payload)-plen))) - if self.master is None: - print("FTP: Can't send request, no master...") - return - self.master.mav.file_transfer_protocol_send(self.network, self.target_system, self.target_component, payload) - self.seq = (self.seq + 1) % 256 - self.last_op = op now = time.time() + if not preserve_seq: + self.seq = (self.seq + 1) % 256 + self.request_retries = 0 + self.send_times[op.seq] = now + else: + # Do not use replies to retransmitted requests as RTT samples: a + # reply may belong to either transmission (Karn's algorithm). + self.send_times[op.seq] = None + self.last_op = op + self.last_op_reply = False if self.ftp_settings.debug > 1: print("> %s dt=%.2f" % (op, now - self.last_op_time)) - self.last_op_time = time.time() + self.last_op_time = now + return payload + + def send(self, op, preserve_seq=False): + '''send a request''' + if self.master is None: + print("FTP: Can't send request, no master...") + return + payload = self.prepare_send(op, preserve_seq=preserve_seq) + self.manager.send_payloads(self, [payload]) + + def send_batch(self, ops): + '''send requests in one link write when supported by pymavlink''' + if len(ops) == 0: + return + if self.master is None: + print("FTP: Can't send request, no master...") + return + payloads = [self.prepare_send(op) for op in ops] + self.manager.send_payloads(self, payloads) + + def update_rtt(self, sample): + '''Update the smoothed RTT and variance from an unambiguous reply.''' + sample = max(0.001, sample) + if not self.rtt_valid: + self.rtt = sample + self.rttvar = sample / 2.0 + self.rtt_valid = True + return + self.rttvar = 0.75 * self.rttvar + 0.25 * abs(self.rtt - sample) + self.rtt = 0.875 * self.rtt + 0.125 * sample + + def retry_timeout(self): + '''Return an RTT-sensitive retransmission timeout.''' + minimum = max(0.05, self.ftp_settings.retry_time) + if not self.rtt_valid: + return max(1.0, minimum) + return max(minimum, min(10.0, self.rtt + 4.0 * self.rttvar)) def terminate_session(self, outcome="failed"): '''terminate current session. outcome describes an incomplete transfer for the status line: "cancelled" when the user or a new command ended it, "failed" for an error''' + if self.done: + return + self.done = True self.transfer_active = False if self.crccmp_dest is not None: print("crccmp: aborted") @@ -249,10 +287,15 @@ def terminate_session(self, outcome="failed"): self.set_progress_status("%s %s %s" % ( "Uploading" if self.write_list is not None else "Downloading", self.filename, outcome)) + # Requests queued by the lag simulator belong to this operation and + # must never run after cancellation or completion. Queue only the + # final TerminateSession after removing them. + self.manager.discard_delayed(self) self.send(FTP_OP(self.seq, self.session, OP_TerminateSession, 0, 0, 0, 0, None)) self.fh = None self.filename = None self.write_list = None + self.write_open = False if self.callback is not None: # tell caller that the transfer failed self.callback(None) @@ -269,12 +312,12 @@ def terminate_session(self, outcome="failed"): self.read_gap_times = {} self.last_read = None self.last_burst_read = None - self.session = (self.session + 1) % 256 self.reached_eof = False self.backlog = 0 self.duplicates = 0 if self.ftp_settings.debug > 0: print("Terminated session") + self.manager.worker_done(self) def cmd_list(self, args): '''list files''' @@ -286,11 +329,50 @@ def cmd_list(self, args): enc_dname = bytearray(dname, 'ascii') self.total_size = 0 self.dir_offset = 0 - op = FTP_OP(self.seq, self.session, OP_ListDirectory, len(enc_dname), 0, 0, self.dir_offset, enc_dname) + self.list_dname = enc_dname + self.list_with_time = ( + self.ftp_settings.list_time != 0 and + self.manager.list_time_supported.get(self.list_target_key()) is not False) + self.send_list_request() + + def list_target_key(self): + '''Return the target whose listing opcode support is being learned.''' + return (self.target_system, self.target_component) + + def send_list_request(self): + '''Request the next page using state owned by this listing worker.''' + opcode = (OP_ListDirectoryWithTime if self.list_with_time + else OP_ListDirectory) + op = FTP_OP(self.seq, self.session, opcode, len(self.list_dname), 0, 0, + self.dir_offset, self.list_dname) self.send(op) + def list_without_time(self): + '''Restart a listing using the opcode understood by older servers.''' + if self.ftp_settings.debug > 0: + print("FTP: no directory listing with time, retrying without") + self.list_with_time = False + self.dir_offset = 0 + self.total_size = 0 + self.send_list_request() + + def list_mtime_str(self, mtime): + '''Format UTC wire time in local time, or '-' when it is unknown.''' + if mtime == 0: + return '-' + try: + return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mtime)) + except (ValueError, OSError): + return str(mtime) + def handle_list_reply(self, op, m): - '''handle OP_ListDirectory reply''' + '''Handle OP_ListDirectory and OP_ListDirectoryWithTime replies.''' + with_time = op.req_opcode == OP_ListDirectoryWithTime + if with_time != self.list_with_time: + # This is a delayed reply to the opcode used before fallback. + return + if with_time and op.opcode == OP_Ack: + self.manager.list_time_supported[self.list_target_key()] = True if op.opcode == OP_Ack: dentries = sorted(op.payload.split(b'\x00')) #print(dentries) @@ -308,28 +390,51 @@ def handle_list_reply(self, op, m): if d[0] == 'D': print(" D %s" % d[1:]) elif d[0] == 'F': - (name, size) = d[1:].split('\t') - size = int(size) + # Names can contain tabs. Size and optional mtime are the + # fields at the end of an entry, so parse from that end. + fields = d[1:].split('\t') + trailing = 2 if with_time else 1 + if len(fields) < trailing + 1: + print(d) + continue + name = '\t'.join(fields[:-trailing]) + try: + size = int(fields[-trailing]) + except ValueError: + print(d) + continue self.total_size += size - print(" %s\t%u" % (name, size)) + if with_time: + try: + mtime = int(fields[-1]) + except ValueError: + mtime = 0 + print(" %s\t%u\t%s" % ( + name, size, self.list_mtime_str(mtime))) + else: + print(" %s\t%u" % (name, size)) else: print(d) # ask for more - more = self.last_op - more.offset = self.dir_offset - self.send(more) + self.send_list_request() elif op.opcode == OP_Nack and len(op.payload) == 1 and op.payload[0] == ERR_EndOfFile: print("Total size %.2f kByte" % (self.total_size / 1024.0)) self.total_size = 0 + self.terminate_session() + elif (with_time and self.dir_offset == 0 and + op.opcode == OP_Nack and len(op.payload) >= 1 and + op.payload[0] in [ERR_Fail, ERR_UnknownCommand]): + self.manager.list_time_supported[self.list_target_key()] = False + self.list_without_time() else: print('LIST: %s' % op) + self.terminate_session() def cmd_get(self, args, callback=None, callback_progress=None): '''get file''' if len(args) == 0: print("Usage: get FILENAME ") return - self.terminate_session("cancelled") fname = args[0] if len(args) > 1: self.filename = args[1] @@ -416,11 +521,6 @@ def write_payload(self, op): def handle_burst_read(self, op, m): '''handle OP_BurstReadFile reply''' - if self.ftp_settings.pkt_loss_tx > 0: - if random.uniform(0,100) < self.ftp_settings.pkt_loss_tx: - if self.ftp_settings.debug > 0: - print("FTP: dropping TX") - return if self.fh is None or self.filename is None: if op.session != self.session: # old session @@ -517,11 +617,11 @@ def handle_reply_read(self, op, m): print("FTP Unexpected read reply") print(op) return - if self.backlog > 0: - self.backlog -= 1 if op.opcode == OP_Ack and self.fh is not None: gap = (op.offset, op.size) if gap in self.read_gaps: + if self.read_gap_times[gap] > 0 and self.backlog > 0: + self.backlog -= 1 self.read_gaps.remove(gap) self.read_gap_times.pop(gap) ofs = self.fh.tell() @@ -598,7 +698,9 @@ def cmd_put(self, args, fh=None, callback=None, progress_callback=None): self.write_idx = 0 self.write_recv_idx = -1 self.write_pending = 0 + self.write_inflight = set() self.write_last_send = None + self.write_open = False self.put_callback = callback self.put_callback_progress = progress_callback @@ -607,7 +709,10 @@ def cmd_put(self, args, fh=None, callback=None, progress_callback=None): self.read_retries = 0 self.op_start = time.time() enc_fname = bytearray(self.filename, 'ascii') - op = FTP_OP(self.seq, self.session, OP_CreateFile, len(enc_fname), 0, 0, 0, enc_fname) + # burst_complete is otherwise unused for CreateFile. It opts in to + # cumulative ACKs from servers that can commit contiguous write + # requests as a batch; older servers safely ignore it. + op = FTP_OP(self.seq, self.session, OP_CreateFile, len(enc_fname), 0, 1, 0, enc_fname) self.send(op) def write_block_len(self, idx): @@ -624,7 +729,9 @@ def put_finished(self, flen): self.put_callback(flen) self.put_callback = None else: - print("Sent file of length ", flen) + dt = max(time.time() - self.op_start, 1.0e-6) + print("Sent file of length %u in %.2fs %.1fkByte/s" % + (flen, dt, (flen / dt) / 1024.0)) self.finished_status("uploading", self.filename, flen) def handle_create_file_reply(self, op, m): @@ -633,6 +740,12 @@ def handle_create_file_reply(self, op, m): self.terminate_session() return if op.opcode == OP_Ack: + self.write_open = True + if op.size >= 6 and bytes(op.payload[:4]) == WRITE_CAPABILITY_MAGIC: + if self.ftp_settings.write_qsize <= 0: + self.write_qsize = max(1, op.payload[4]) + if self.ftp_settings.write_batch_size <= 0: + self.write_batch_size = max(1, op.payload[5]) self.send_more_writes() else: print("Create failed") @@ -640,6 +753,8 @@ def handle_create_file_reply(self, op, m): def send_more_writes(self): '''send some more writes''' + if not self.write_open: + return if len(self.write_list) == 0: # all done self.put_finished(self.write_file_size) @@ -648,24 +763,36 @@ def send_more_writes(self): now = time.time() if self.write_last_send is not None: - if now - self.write_last_send > max(min(10*self.rtt, 1),0.2): + if now - self.write_last_send > self.retry_timeout(): # we seem to have lost a block of replies - self.write_pending = max(0, self.write_pending-1) + self.write_inflight.clear() + self.write_pending = 0 + self.write_last_send = now + + qsize = max(1, self.write_qsize) + batch_size = max(1, min(self.write_batch_size, qsize)) + free_slots = qsize - self.write_pending + if self.write_pending > 0 and free_slots < batch_size: + return - n = min(self.ftp_settings.write_qsize-self.write_pending, len(self.write_list)) + unsent = len(self.write_list - self.write_inflight) + n = min(free_slots, unsent) + writes = [] for i in range(n): # send in round-robin, skipping any that have been acked idx = self.write_idx - while idx not in self.write_list: + while idx not in self.write_list or idx in self.write_inflight: idx = (idx + 1) % self.write_total ofs = idx * self.write_block_size self.fh.seek(ofs) data = self.fh.read(self.write_block_size) write = FTP_OP(self.seq, self.session, OP_WriteFile, len(data), 0, 0, ofs, bytearray(data)) - self.send(write) + writes.append(write) self.write_idx = (idx + 1) % self.write_total + self.write_inflight.add(idx) self.write_pending += 1 self.write_last_send = now + self.send_batch(writes) def handle_write_reply(self, op, m): '''handle OP_WriteFile reply''' @@ -673,24 +800,53 @@ def handle_write_reply(self, op, m): self.terminate_session() return if op.opcode != OP_Ack: - print("Write failed") + print("Write failed: %s" % op) self.terminate_session() return - # assume the FTP server processes the blocks sequentially. This means - # when we receive an ack that any blocks between the last ack and this - # one have been lost + # Legacy servers ACK one block at a time. Negotiated batch ACKs carry + # an explicit contiguous range so gaps remain eligible for retry. idx = op.offset // self.write_block_size - count = (idx - self.write_recv_idx) % self.write_total - - self.write_pending = max(0, self.write_pending - count) + previous_idx = self.write_recv_idx + acked = [idx] + if op.burst_complete: + if op.size >= 5 and self.ftp_settings.write_qsize <= 0: + # The server may enlarge this session's reservation when a + # competing writer closes. Windows never shrink while + # requests are outstanding. + self.write_qsize = max(self.write_qsize, op.payload[4]) + if op.size >= 4: + start_offset, = struct.unpack(' 1: print("< %s dt=%.2f" % (op, dt)) self.last_op_time = now - if self.ftp_settings.pkt_loss_rx > 0: - if random.uniform(0,100) < self.ftp_settings.pkt_loss_rx: - if self.ftp_settings.debug > 1: - print("FTP: dropping packet RX") - return - if op.req_opcode == self.last_op.opcode and op.seq == (self.last_op.seq + 1) % 256: - self.rtt = max(min(self.rtt, dt), 0.01) - if op.req_opcode == OP_ListDirectory: + request_seq = (op.seq - 1) % 256 + sent = self.send_times.pop(request_seq, None) + if sent is not None: + self.update_rtt(now - sent) + + if op.opcode == OP_Nack and op.payload is not None and \ + len(op.payload) == 1 and op.payload[0] == ERR_NoSessionsAvailable: + # Another client may also be using the server, so the local + # concurrency cap is not sufficient on its own. Keep this + # operation intact and retry instead of failing its callback. + self.session_waiting = True + self.last_op_time = now + if not self.session_wait_reported: + print("FTP: no sessions available, waiting to retry") + self.session_wait_reported = True + return + + if self.last_op is not None and \ + op.req_opcode == self.last_op.opcode and \ + op.seq == (self.last_op.seq + 1) % 256: + self.last_op_reply = True + if op.req_opcode in [OP_ListDirectory, + OP_ListDirectoryWithTime]: self.handle_list_reply(op, m) elif op.req_opcode == OP_OpenFileRO: self.handle_open_RO_reply(op, m) @@ -1073,45 +1249,81 @@ def send_gap_read(self, g): print("Gap read of %u at %u rem=%u blog=%u" % (length, offset, len(self.read_gaps), self.backlog)) read = FTP_OP(self.seq, self.session, OP_ReadFile, length, 0, 0, offset, None) self.send(read) - self.read_gaps.remove(g) - self.read_gaps.append(g) self.last_gap_send = time.time() self.read_gap_times[g] = self.last_gap_send self.backlog += 1 def check_read_send(self): - '''see if we should send another gap read''' + '''keep a bounded window of gap reads in flight''' if len(self.read_gaps) == 0: return - g = self.read_gaps[0] now = time.time() - dt = now - self.read_gap_times[g] - if not self.reached_eof: - # send gap reads once - for g in self.read_gap_times.keys(): - if self.read_gap_times[g] == 0: - self.send_gap_read(g) - return - if self.read_gap_times[g] > 0 and dt > self.ftp_settings.retry_time: - if self.backlog > 0: - self.backlog -= 1 - self.read_gap_times[g] = 0 - - if self.read_gap_times[g] != 0: - # still pending - return - if not self.reached_eof and self.backlog >= self.ftp_settings.max_backlog: - # don't fill queue too far until we have got past the burst - return - if now - self.last_gap_send < 0.05: - # don't send too fast - return - self.send_gap_read(g) + timeout = self.retry_timeout() + for g in self.read_gaps: + sent = self.read_gap_times[g] + if sent > 0 and now - sent > timeout: + self.read_gap_times[g] = 0 + if self.backlog > 0: + self.backlog -= 1 + + limit = max(1, self.ftp_settings.max_backlog) + for g in self.read_gaps: + if self.backlog >= limit: + break + if self.read_gap_times[g] == 0: + self.send_gap_read(g) def idle_task(self): '''check for file gaps and lost requests''' now = time.time() + if self.session_waiting: + if now - self.last_op_time >= self.retry_timeout(): + self.session_waiting = False + if self.last_op.opcode == OP_OpenFileRO: + self.op_start = now + self.send(self.last_op) + return + + # ArduPilot's incoming FTP request queue has the same depth as its + # session table. Under contention an initial request can therefore be + # dropped before the server has a chance to NACK it. Reusing the same + # sequence number makes this safe whether the request or its reply was + # lost: the server's duplicate-request cache returns the old reply. + initial_opcodes = ( + OP_ListDirectory, OP_ListDirectoryWithTime, OP_OpenFileRO, + OP_CreateFile, OP_RemoveFile, OP_RemoveDirectory, OP_Rename, + OP_CreateDirectory, OP_CalcFileCRC32, + ) + initial_timeout = self.retry_timeout() + if self.last_op is not None and \ + self.last_op.opcode == OP_ListDirectoryWithTime: + # Capability probing needs a generous floor: variable-lag links + # should not make a capable server look like an old one. + initial_timeout = max( + initial_timeout, self.ftp_settings.list_time_timeout) + if self.last_op is not None and not self.last_op_reply and \ + self.last_op.opcode in initial_opcodes and \ + now - self.last_op_time > initial_timeout: + if self.last_op.opcode == OP_ListDirectoryWithTime and \ + self.dir_offset == 0 and \ + self.request_retries >= self.ftp_settings.list_retries: + # Some servers silently ignore unknown FTP opcodes. After + # several RTT-sensitive retries, remember the old server and + # restart this listing using the baseline opcode. + self.manager.list_time_supported[self.list_target_key()] = False + self.list_without_time() + return + self.request_retries += 1 + if self.request_retries > 10: + print("FTP: request timed out: %s" % self.last_op) + self.terminate_session() + return + if self.ftp_settings.debug > 0: + print("FTP: retry request: %s" % self.last_op) + self.send(self.last_op, preserve_seq=True) + return + # ahead of the early returns below, which skip idle transfers self.update_status() @@ -1122,23 +1334,6 @@ def idle_task(self): self.crccmp_expect = None self.crccmp_next() - # see if we lost an open reply - if self.op_start is not None and now - self.op_start > 1.0 and self.last_op.opcode == OP_OpenFileRO: - self.op_start = now - self.open_retries += 1 - if self.open_retries > 2: - # fail the get - self.op_start = None - self.terminate_session() - return - if self.ftp_settings.debug > 0: - print("FTP: retry open") - send_op = self.last_op - self.send(FTP_OP(self.seq, self.session, OP_TerminateSession, 0, 0, 0, 0, None)) - self.session = (self.session + 1) % 256 - send_op.session = self.session - self.send(send_op) - if len(self.read_gaps) == 0 and self.last_burst_read is None and self.write_list is None: return @@ -1146,7 +1341,8 @@ def idle_task(self): return # see if burst read has stalled - if not self.reached_eof and self.last_burst_read is not None and now - self.last_burst_read > self.ftp_settings.retry_time: + if not self.reached_eof and self.last_burst_read is not None and \ + now - self.last_burst_read > self.retry_timeout(): dt = now - self.last_burst_read self.last_burst_read = now if self.ftp_settings.debug > 0: @@ -1160,6 +1356,407 @@ def idle_task(self): if self.write_list is not None: self.send_more_writes() + +class FTPModule(mp_module.MPModule): + '''Public FTP module and concurrent-session manager.''' + + def __init__(self, mpstate): + super(FTPModule, self).__init__(mpstate, "ftp", public=True) + self.add_command('ftp', self.cmd_ftp, "file transfer", + ["", + "set (FTPSETTING)", + "put (FILENAME) (FILENAME)", + "crclocal (FILENAME)", + "crccmp (FILENAME)"]) + self.ftp_settings = mp_settings.MPSettings( + [('debug', int, 0), + ('pkt_loss_tx', int, 0), + ('pkt_loss_rx', int, 0), + ('pkt_lag_tx', float, 0.0), + ('pkt_lag_rx', float, 0.0), + ('pkt_lag_jitter_tx', float, 0.0), + ('pkt_lag_jitter_rx', float, 0.0), + ('loss_seed', int, 0), + ('max_backlog', int, 5), + ('burst_read_size', int, MAX_Payload), + ('write_size', int, MAX_Payload), + # zero selects the server-advertised values, with conservative + # fallbacks for servers predating write batching + ('write_qsize', int, 0), + ('write_batch_size', int, 0), + ('retry_time', float, 0.5), + ('crccmp_timeout', float, 120.0), + ('list_time', int, 1), + ('list_time_timeout', float, 3.0), + ('list_retries', int, 3), + # ArduPilot currently has five GCS_FTP server sessions. Keeping + # the cap configurable also supports smaller/custom servers. + ('max_sessions', int, 5)]) + self.add_completion_function('(FTPSETTING)', + self.ftp_settings.completion) + self.workers = {} + self.pending = [] + # Cache ListDirectoryWithTime support independently for each target. + self.list_time_supported = {} + # A previous process can leave delayed packets or a cached reply on a + # poor link. Starting every process at session zero can then turn a + # stale CreateFile ACK into writes against a closed server session. + self.next_session = random.SystemRandom().randrange(256) + self.warned_component = False + self.loss_rng = random.Random() + self.active_loss_seed = None + self.delay_sequence = 0 + self.tx_delay_queue = [] + self.rx_delay_queue = [] + self.last_tx_deadline = 0.0 + self.last_rx_deadline = 0.0 + self.retired_sessions = {} + + def packet_lost(self, direction): + '''Return true when the configured link simulator drops a packet.''' + seed = self.ftp_settings.loss_seed + if seed != self.active_loss_seed: + self.loss_rng.seed(None if seed == 0 else seed) + self.active_loss_seed = seed + percent = (self.ftp_settings.pkt_loss_tx if direction == 'TX' + else self.ftp_settings.pkt_loss_rx) + lost = percent > 0 and self.loss_rng.uniform(0, 100) < percent + if lost and self.ftp_settings.debug > 1: + print("FTP: dropping packet %s" % direction) + return lost + + def packet_delay(self, direction): + '''Return simulated one-way delay in seconds. + + Jitter is a uniformly distributed extra delay. Delivery deadlines are + constrained separately per direction so jitter models FIFO + head-of-line blocking instead of reordering a serial telemetry link. + ''' + if direction == 'TX': + base = self.ftp_settings.pkt_lag_tx + jitter = self.ftp_settings.pkt_lag_jitter_tx + else: + base = self.ftp_settings.pkt_lag_rx + jitter = self.ftp_settings.pkt_lag_jitter_rx + delay_ms = max(0.0, base) + if jitter > 0: + delay_ms += self.loss_rng.uniform(0, jitter) + return delay_ms * 0.001 + + def _transmit_payloads(self, master, network, target_system, + target_component, payloads): + '''Serialize and transmit one logical batch of FTP requests.''' + mav = master.mav + if len(payloads) <= 1 or not hasattr(mav, 'file'): + for payload in payloads: + mav.file_transfer_protocol_send( + network, target_system, target_component, payload) + return + + link = mav.file + collector = MAVLinkBatchWriter() + mav.file = collector + try: + for payload in payloads: + mav.file_transfer_protocol_send( + network, target_system, target_component, payload) + finally: + mav.file = link + if not collector.packets: + return + + port = getattr(link, 'port', None) + port_type = getattr(port, 'type', None) + link_name = type(link).__name__ + is_stream = (link_name in ('mavtcp', 'mavtcpin') or + port_type == socket.SOCK_STREAM) + is_network = (link_name == 'mavudp' or + port_type == socket.SOCK_DGRAM or is_stream) + if not is_network: + # Serial writes benefit substantially from being combined into a + # single USB transfer. + self._write_link_data(link, b''.join(collector.packets), False) + return + + batch = bytearray() + for packet in collector.packets: + if batch and len(batch) + len(packet) > MAX_NETWORK_BATCH: + self._write_link_data(link, bytes(batch), is_stream) + batch = bytearray() + batch.extend(packet) + if batch: + self._write_link_data(link, bytes(batch), is_stream) + + def _write_link_data(self, link, data, is_stream): + '''Write one encoded batch without ignoring partial stream writes.''' + if is_stream: + port = getattr(link, 'port', None) + if port is None and hasattr(link, 'reconnect'): + try: + link.reconnect() + except OSError: + return + port = getattr(link, 'port', None) + if port is not None and hasattr(port, 'sendall'): + try: + port.sendall(data) + except OSError: + if hasattr(link, 'handle_disconnect'): + link.handle_disconnect() + return + + offset = 0 + while offset < len(data): + written = link.write(data[offset:]) + # Datagram and several pymavlink wrappers return None after a + # complete write. Integer-returning serial writers can be safely + # resumed when they accept only part of the buffer. + if written is None: + return + if written <= 0: + return + offset += written + + def send_payloads(self, worker, payloads): + '''Apply outgoing loss/lag, preserving batches that survive.''' + payloads = [bytes(payload) for payload in payloads + if not self.packet_lost('TX')] + if not payloads: + return + args = (worker.master, worker.network, worker.target_system, + worker.target_component, payloads) + lag = self.packet_delay('TX') + if lag == 0: + self._transmit_payloads(*args) + return + self.delay_sequence += 1 + deadline = max(time.monotonic() + lag, self.last_tx_deadline) + self.last_tx_deadline = deadline + heapq.heappush(self.tx_delay_queue, + (deadline, self.delay_sequence, worker, args)) + + def _packet_worker(self, m): + try: + session = m.payload[2] + except (IndexError, TypeError): + return None + return self.workers.get(session) + + def cmd_ftp(self, args): + '''FTP operations''' + usage = "Usage: ftp " + if len(args) < 1: + print(usage) + return + command = args[0] + if command == 'set': + self.ftp_settings.command(args[1:]) + elif command == 'status': + self.cmd_status() + elif command == 'cancel': + self.cmd_cancel() + elif command == 'crclocal': + self.cmd_crclocal(args[1:]) + else: + method = getattr(self, 'cmd_' + command, None) + if method is None: + print(usage) + else: + method(args[1:]) + + def _allocate_session(self): + '''Return an unused client-selected uint8 session id.''' + now = time.monotonic() + self.retired_sessions = { + session: deadline + for session, deadline in self.retired_sessions.items() + if deadline > now + } + for _ in range(256): + session = self.next_session + self.next_session = (self.next_session + 1) % 256 + if session not in self.workers and \ + session not in self.retired_sessions: + return session + return None + + def _launch(self, operation): + session = self._allocate_session() + if session is None: + self.pending.insert(0, operation) + return None + worker = FTPWorker(self, session) + worker.operation_name = operation['name'] + self.workers[session] = worker + method = getattr(worker, operation['method']) + method(*operation['args'], **operation['kwargs']) + # Bad arguments or a local-file error can return without sending. + if worker.last_op is None: + self.worker_done(worker) + return worker + + def _submit(self, name, method, *args, **kwargs): + operation = { + 'name': name, + 'method': method, + 'args': args, + 'kwargs': kwargs, + } + limit = self._session_limit() + if len(self.workers) >= limit: + self.pending.append(operation) + print("FTP: queued %s (%u sessions active)" % + (name, len(self.workers))) + return None + return self._launch(operation) + + def worker_done(self, worker): + '''Forget a completed worker and start the oldest queued operation.''' + if self.workers.get(worker.session) is worker: + del self.workers[worker.session] + deadline = time.monotonic() + SESSION_REUSE_DELAY + # A configured lag can exceed the normal network quarantine. Do + # not reuse the ID before its delayed TerminateSession is sent. + for queued_deadline, _, queued_worker, _ in self.tx_delay_queue: + if queued_worker is worker: + deadline = max(deadline, queued_deadline + 1.0) + self.retired_sessions[worker.session] = deadline + self._start_pending() + + def _start_pending(self): + '''Start queued work when both a slot and a safe session ID exist.''' + limit = self._session_limit() + while self.pending and len(self.workers) < limit: + operation = self.pending.pop(0) + if self._launch(operation) is None: + break + + def discard_delayed(self, worker): + '''Discard simulated-link traffic belonging to a finished worker.''' + self.tx_delay_queue = [ + item for item in self.tx_delay_queue if item[2] is not worker + ] + heapq.heapify(self.tx_delay_queue) + self.rx_delay_queue = [ + item for item in self.rx_delay_queue if item[2] is not worker + ] + heapq.heapify(self.rx_delay_queue) + + def _session_limit(self): + # Session ids are uint8. Keep one value in reserve so allocation and + # queuing remain well defined even with an accidental oversized setting. + return min(255, max(1, int(self.ftp_settings.max_sessions))) + + def cmd_list(self, args): + return self._submit('list', 'cmd_list', args) + + def cmd_get(self, args, callback=None, callback_progress=None): + return self._submit('get', 'cmd_get', args, + callback=callback, + callback_progress=callback_progress) + + def cmd_put(self, args, fh=None, callback=None, progress_callback=None): + return self._submit('put', 'cmd_put', args, fh=fh, + callback=callback, + progress_callback=progress_callback) + + def cmd_rm(self, args): + return self._submit('rm', 'cmd_rm', args) + + def cmd_rmdir(self, args): + return self._submit('rmdir', 'cmd_rmdir', args) + + def cmd_rename(self, args): + return self._submit('rename', 'cmd_rename', args) + + def cmd_mkdir(self, args): + return self._submit('mkdir', 'cmd_mkdir', args) + + def cmd_crc(self, args): + return self._submit('crc', 'cmd_crc', args) + + def cmd_crccmp(self, args): + return self._submit('crccmp', 'cmd_crccmp', args) + + def cmd_crclocal(self, args): + # This operation is entirely local and consumes no server session. + return FTPWorker(self, 0).cmd_crclocal(args) + + def cmd_cancel(self): + '''Cancel all active and queued operations.''' + pending = self.pending + self.pending = [] + for operation in pending: + callback = operation['kwargs'].get('callback') + if callback is not None: + callback(None) + progress = operation['kwargs'].get('progress_callback') + if progress is not None: + progress(None) + for worker in list(self.workers.values()): + worker.terminate_session("cancelled") + + def cmd_status(self): + if not self.workers and not self.pending: + # Keep the longstanding wording used by scripts and autotests. + print("No transfer in progress") + return + for session, worker in sorted(self.workers.items()): + status = worker.transfer_status() + if status is None: + status = worker.operation_name + if worker.session_waiting: + status += " (waiting for a server session)" + print("FTP session %u: %s" % (session, status)) + if self.pending: + print("FTP queued: %s" % + ', '.join(operation['name'] for operation in self.pending)) + + def mavlink_packet(self, m): + if m.get_type() != "FILE_TRANSFER_PROTOCOL": + return + if (m.target_system != self.settings.source_system or + m.target_component != self.settings.source_component): + if m.target_system == self.settings.source_system and not self.warned_component: + self.warned_component = True + print("FTP reply for mavlink component %u" % m.target_component) + return + if self.packet_lost('RX'): + return + worker = self._packet_worker(m) + if worker is None: + return + lag = self.packet_delay('RX') + if lag == 0: + worker.mavlink_packet(m) + return + self.delay_sequence += 1 + deadline = max(time.monotonic() + lag, self.last_rx_deadline) + self.last_rx_deadline = deadline + heapq.heappush(self.rx_delay_queue, + (deadline, self.delay_sequence, worker, m)) + + def idle_task(self): + now = time.monotonic() + while self.tx_delay_queue and self.tx_delay_queue[0][0] <= now: + _, _, worker, args = heapq.heappop(self.tx_delay_queue) + active = self.workers.get(worker.session) is worker + terminal = all(payload[3] == OP_TerminateSession + for payload in args[-1]) + if active or terminal: + self._transmit_payloads(*args) + while self.rx_delay_queue and self.rx_delay_queue[0][0] <= now: + _, _, worker, m = heapq.heappop(self.rx_delay_queue) + if self.workers.get(worker.session) is worker: + worker.mavlink_packet(m) + for worker in list(self.workers.values()): + worker.idle_task() + self._start_pending() + + def unload(self): + self.cmd_cancel() + super(FTPModule, self).unload() + def init(mpstate): '''initialise module''' return FTPModule(mpstate) diff --git a/tests/test_mavproxy_ftp.py b/tests/test_mavproxy_ftp.py new file mode 100644 index 0000000000..55ff352c5b --- /dev/null +++ b/tests/test_mavproxy_ftp.py @@ -0,0 +1,574 @@ +import importlib.util +import contextlib +import io +from pathlib import Path +import struct +import socket +import time +import types +import unittest + + +# Some developer environments have a released MAVProxy imported by a pytest +# plugin before collection begins. Load the worktree file explicitly so the +# tests always exercise the code they accompany. +FTP_MODULE_PATH = (Path(__file__).resolve().parents[1] / + 'MAVProxy/modules/mavproxy_ftp.py') +FTP_SPEC = importlib.util.spec_from_file_location( + 'mavproxy_ftp_under_test', FTP_MODULE_PATH) +mavproxy_ftp = importlib.util.module_from_spec(FTP_SPEC) +FTP_SPEC.loader.exec_module(mavproxy_ftp) + + +class FakeMAV: + def __init__(self): + self.sent = [] + + def file_transfer_protocol_send(self, network, target_system, + target_component, payload): + self.sent.append(bytes(payload)) + + +class FakeMaster: + def __init__(self): + self.mav = FakeMAV() + + +class FakeLink: + def __init__(self, port_type=None): + self.writes = [] + self.port = types.SimpleNamespace(type=port_type) + + def write(self, data): + self.writes.append(bytes(data)) + return len(data) + + +class FakeStreamPort: + type = socket.SOCK_STREAM + + def __init__(self): + self.writes = [] + + def sendall(self, data): + self.writes.append(bytes(data)) + + +class FakeBatchMAV: + '''Minimal MAVLink encoder whose output can be collected by the module.''' + def __init__(self, link): + self.file = link + + def file_transfer_protocol_send(self, network, target_system, + target_component, payload): + self.file.write(b'F' + bytes(payload)) + + +class FakeConsole: + def __init__(self): + self.status = {} + + def set_status(self, name, value, row=None): + self.status[name] = (value, row) + + +class FakeMPState: + def __init__(self): + self.public_modules = {} + self.command_map = {} + self.completions = {} + self.completion_functions = {} + self.settings = types.SimpleNamespace( + target_system=1, + target_component=1, + source_system=255, + source_component=0, + ) + self.console = FakeConsole() + self._master = FakeMaster() + + def master(self): + return self._master + + +class FTPMessage: + def __init__(self, payload): + self.payload = payload + self.target_system = 255 + self.target_component = 0 + + def get_type(self): + return "FILE_TRANSFER_PROTOCOL" + + +def reply(session, req_opcode, opcode=mavproxy_ftp.OP_Ack, payload=b'', + offset=0, burst_complete=0, seq=1): + header = struct.pack( + '