From 0e87e0c7de839a11e4bb31a9ff1158615a78d0fa Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Tue, 8 Aug 2023 06:28:43 +0530 Subject: [PATCH] Work on refactoring send kitten to fix various issues --- kittens/clipboard/legacy.go | 2 +- kittens/clipboard/write.go | 4 +- kittens/transfer/receive.go | 2 +- kittens/transfer/send.go | 116 +++++++++++++++++------------------- tools/cmd/at/tty_io.go | 4 +- tools/tui/loop/api.go | 2 +- tools/tui/loop/run.go | 3 +- tools/tui/loop/write.go | 8 +-- 8 files changed, 69 insertions(+), 72 deletions(-) diff --git a/kittens/clipboard/legacy.go b/kittens/clipboard/legacy.go index 74fec7dc0..58858b939 100644 --- a/kittens/clipboard/legacy.go +++ b/kittens/clipboard/legacy.go @@ -164,7 +164,7 @@ func run_plain_text_loop(opts *Options) (err error) { return "", nil } - lp.OnWriteComplete = func(id loop.IdType) error { + lp.OnWriteComplete = func(id loop.IdType, has_pending_writes bool) error { if id == enc_writer.last_written_id { return write_one_chunk() } diff --git a/kittens/clipboard/write.go b/kittens/clipboard/write.go index 8fdf3a1dd..c42c667a7 100644 --- a/kittens/clipboard/write.go +++ b/kittens/clipboard/write.go @@ -111,14 +111,14 @@ func write_loop(inputs []*Input, opts *Options) (err error) { lp.QueueWriteString(encode(make_metadata("wdata", ""), "")) waiting_for_write = 0 } - return lp.OnWriteComplete(waiting_for_write) + return lp.OnWriteComplete(waiting_for_write, false) } return fmt.Errorf("Failed to read from %s with error: %w", i.arg, err) } return nil } - lp.OnWriteComplete = func(msg_id loop.IdType) error { + lp.OnWriteComplete = func(msg_id loop.IdType, has_pending_writes bool) error { if waiting_for_write == msg_id { return write_chunk() } diff --git a/kittens/transfer/receive.go b/kittens/transfer/receive.go index f0ee3d95d..a4b387c21 100644 --- a/kittens/transfer/receive.go +++ b/kittens/transfer/receive.go @@ -881,7 +881,7 @@ func (self *handler) on_file_transfer_response(ftc *FileTransmissionCommand) (er return } -func (self *handler) on_writing_finished(msg_id loop.IdType) (err error) { +func (self *handler) on_writing_finished(msg_id loop.IdType, has_pending_writes bool) (err error) { if self.quit_after_write_code > -1 { self.lp.Quit(self.quit_after_write_code) } else if msg_id == self.last_data_write_id { diff --git a/kittens/transfer/send.go b/kittens/transfer/send.go index e547ac284..38ef75ffa 100644 --- a/kittens/transfer/send.go +++ b/kittens/transfer/send.go @@ -316,10 +316,8 @@ func (self *ProgressTracker) start_transfer() { self.started_at = t.at } -func (self *ProgressTracker) on_transmit(amt int64) { - if self.active_file != nil { - self.active_file.transmitted_bytes += amt - } +func (self *ProgressTracker) on_transmit(amt int64, active_file *File) { + active_file.transmitted_bytes += amt self.total_transferred += amt now := time.Now() self.transfers = append(self.transfers, &Transfer{amt: amt, at: now}) @@ -359,6 +357,7 @@ type SendManager struct { progress_tracker ProgressTracker current_chunk_uncompressed_sz int64 current_chunk_write_id loop.IdType + current_chunk_for_file_id string } func (self *SendManager) start_transfer() string { @@ -381,6 +380,7 @@ func (self *SendManager) initialize() { } self.active_idx = -1 self.current_chunk_uncompressed_sz = -1 + self.current_chunk_for_file_id = "" self.prefix = fmt.Sprintf("\x1b]%d;id=%s;", kitty.FileTransferCode, self.request_id) self.suffix = "\x1b\\" for _, f := range self.files { @@ -552,10 +552,12 @@ func (self *SendHandler) draw_progress() { } else { af := self.manager.last_progress_file if af == nil || self.done_file_ids.Has(af.file_id) { - if self.manager.has_rsync && !self.manager.has_transmitting { - self.lp.QueueWriteString(sc + ` Transferring rsync signatures...`) - } else { - self.lp.QueueWriteString(sc + ` Transferring metadata...`) + if !self.manager.has_transmitting && self.done_file_ids.Len() == 0 { + if self.manager.has_rsync { + self.lp.QueueWriteString(sc + ` Transferring rsync signatures...`) + } else { + self.lp.QueueWriteString(sc + ` Transferring metadata...`) + } } } else { self.draw_progress_for_current_file(af, sc, false) @@ -633,7 +635,7 @@ func (self *SendHandler) on_file_done(f *File) { if f.err_msg != "" { self.failed_files = append(self.failed_files, f) } - self.schedule_progress_update(100 * time.Millisecond) + self.refresh_progress(0) } func (self *SendHandler) send_payload(payload string) loop.IdType { @@ -827,25 +829,13 @@ func (self *SendHandler) on_file_transfer_response(ftc *FileTransmissionCommand) self.send_file_metadata() } } - self.lp.CallSoon(self.loop_tick) - return nil -} - -func (self *SendHandler) loop_tick(timer_id loop.IdType) (err error) { - if self.manager.state == SEND_WAITING_FOR_PERMISSION { - return - } - if self.transmit_started { - if err = self.transmit_next_chunk(); err != nil { - return err - } - if err = self.refresh_progress(timer_id); err != nil { - return err - } - } else { + if !self.transmit_started { return self.check_for_transmit_ok() } - return + if self.manager.all_acknowledged { + self.transfer_finished() + } + return nil } func (self *SendHandler) check_for_transmit_ok() (err error) { @@ -982,39 +972,39 @@ func (self *File) next_chunk() (ans string, asz int, err error) { } func (self *SendManager) next_chunks(callback func(string) loop.IdType) error { - for { - if self.active_file() == nil { - self.activate_next_ready_file() + if self.active_file() == nil { + self.activate_next_ready_file() + } + af := self.active_file() + if af == nil { + return nil + } + chunk := "" + self.current_chunk_uncompressed_sz = 0 + for af.state != FINISHED && len(chunk) == 0 { + c, usz, err := af.next_chunk() + if err != nil { + return err } - af := self.active_file() - if af == nil { + self.current_chunk_uncompressed_sz += int64(usz) + self.current_chunk_for_file_id = af.file_id + chunk = c + } + is_last := af.state == FINISHED + if len(chunk) > 0 { + split_for_transfer(utils.UnsafeStringToBytes(chunk), af.file_id, is_last, func(ftc *FileTransmissionCommand) { + self.current_chunk_write_id = callback(ftc.Serialize()) + }) + } else if is_last { + self.current_chunk_write_id = callback(FileTransmissionCommand{Action: Action_end_data, File_id: af.file_id}.Serialize()) + } + if is_last { + self.activate_next_ready_file() + if self.active_file() == nil { return nil } - chunk := "" - self.current_chunk_uncompressed_sz = 0 - for af.state != FINISHED && len(chunk) == 0 { - c, usz, err := af.next_chunk() - if err != nil { - return err - } - self.current_chunk_uncompressed_sz += int64(usz) - chunk = c - } - is_last := af.state == FINISHED - if len(chunk) > 0 { - split_for_transfer(utils.UnsafeStringToBytes(chunk), af.file_id, is_last, func(ftc *FileTransmissionCommand) { - self.current_chunk_write_id = callback(ftc.Serialize()) - }) - } else if is_last { - self.current_chunk_write_id = callback(FileTransmissionCommand{Action: Action_end_data, File_id: af.file_id}.Serialize()) - } - if is_last { - self.activate_next_ready_file() - if self.active_file() == nil { - return nil - } - } } + return nil } func (self *SendHandler) transmit_next_chunk() (err error) { @@ -1142,12 +1132,13 @@ func (self *SendHandler) on_key_event(ev *loop.KeyEvent) error { return nil } -func (self *SendHandler) on_writing_finished(msg_id loop.IdType) (err error) { +func (self *SendHandler) on_writing_finished(msg_id loop.IdType, has_pending_writes bool) (err error) { chunk_transmitted := self.manager.current_chunk_uncompressed_sz >= 0 && msg_id == self.manager.current_chunk_write_id if chunk_transmitted { - self.manager.progress_tracker.on_transmit(self.manager.current_chunk_uncompressed_sz) + self.manager.progress_tracker.on_transmit(self.manager.current_chunk_uncompressed_sz, self.manager.fid_map[self.manager.current_chunk_for_file_id]) self.manager.current_chunk_uncompressed_sz = -1 self.manager.current_chunk_write_id = 0 + self.manager.current_chunk_for_file_id = "" } if self.finish_cmd_write_id > 0 && msg_id == self.finish_cmd_write_id { if len(self.failed_files) > 0 { @@ -1155,13 +1146,18 @@ func (self *SendHandler) on_writing_finished(msg_id loop.IdType) (err error) { } else { self.quit_after_write_code = 0 } + self.refresh_progress(0) } - if self.quit_after_write_code > -1 { + if self.quit_after_write_code > -1 && !has_pending_writes { self.lp.Quit(self.quit_after_write_code) return } - if self.manager.state == SEND_PERMISSION_GRANTED && (!self.transmit_started || chunk_transmitted) { - self.lp.CallSoon(self.loop_tick) + if self.manager.state == SEND_PERMISSION_GRANTED && !self.transmit_started { + return self.check_for_transmit_ok() + } + if chunk_transmitted { + self.refresh_progress(0) + return self.transmit_next_chunk() } return } diff --git a/tools/cmd/at/tty_io.go b/tools/cmd/at/tty_io.go index 994ce9bb2..2cbadcab7 100644 --- a/tools/cmd/at/tty_io.go +++ b/tools/cmd/at/tty_io.go @@ -97,7 +97,7 @@ func do_chunked_io(io_data *rc_io_data) (serialized_response []byte, err error) return "", nil } - lp.OnWriteComplete = func(completed_write_id loop.IdType) error { + lp.OnWriteComplete = func(completed_write_id loop.IdType, has_pending_writes bool) error { if state == WAITING_FOR_STREAMING_RESPONSE || state == WAITING_FOR_RESPONSE { return nil } @@ -151,7 +151,7 @@ func do_chunked_io(io_data *rc_io_data) (serialized_response []byte, err error) lp.OnRCResponse = func(raw []byte) error { if state == WAITING_FOR_STREAMING_RESPONSE && is_stream_response(raw) { state = SENDING - return lp.OnWriteComplete(0) + return lp.OnWriteComplete(0, false) } serialized_response = raw lp.Quit(0) diff --git a/tools/tui/loop/api.go b/tools/tui/loop/api.go index d0623aa05..cfb158925 100644 --- a/tools/tui/loop/api.go +++ b/tools/tui/loop/api.go @@ -83,7 +83,7 @@ type Loop struct { OnResize func(old_size ScreenSize, new_size ScreenSize) error // Called when writing is done - OnWriteComplete func(msg_id IdType) error + OnWriteComplete func(msg_id IdType, has_pending_writes bool) error // Called when a response to an rc command is received OnRCResponse func(data []byte) error diff --git a/tools/tui/loop/run.go b/tools/tui/loop/run.go index 780e1b898..3063c7644 100644 --- a/tools/tui/loop/run.go +++ b/tools/tui/loop/run.go @@ -285,6 +285,7 @@ func (self *Loop) run() (err error) { // queueing and startup of writer thread and also as a performance // optimization to avoid copying unnecessarily to pending_writes self.tty_write_channel = make(chan write_msg, 512) + self.write_msg_id_counter = 0 write_done_channel := make(chan IdType) self.wakeup_channel = make(chan byte, 256) self.pending_writes = make([]write_msg, 0, 256) @@ -452,7 +453,7 @@ func (self *Loop) run() (err error) { case msg_id := <-write_done_channel: self.flush_pending_writes(self.tty_write_channel) if self.OnWriteComplete != nil { - err = self.OnWriteComplete(msg_id) + err = self.OnWriteComplete(msg_id, msg_id < self.write_msg_id_counter) if err != nil { return err } diff --git a/tools/tui/loop/write.go b/tools/tui/loop/write.go index 9013fe0a3..a28eb0842 100644 --- a/tools/tui/loop/write.go +++ b/tools/tui/loop/write.go @@ -75,15 +75,15 @@ func (self *Loop) wait_for_write_to_complete(sentinel IdType, tty_write_channel case tty_write_channel <- self.pending_writes[0]: self.pending_writes = self.pending_writes[1:] case write_id, more := <-write_done_channel: - if write_id == sentinel { - return nil - } if self.OnWriteComplete != nil { - err := self.OnWriteComplete(write_id) + err := self.OnWriteComplete(write_id, write_id < self.write_msg_id_counter) if err != nil { return err } } + if write_id == sentinel { + return nil + } if !more { return fmt.Errorf("The write_done_channel was unexpectedly closed") }