mirrors/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty synced 2026-07-06 16:05:05 +02:00
Code Issues Packages Projects Releases Wiki Activity

Work on refactoring send kitten to fix various issues

Browse Source
This commit is contained in:
Kovid Goyal
2023-08-08 06:28:43 +05:30
parent 0971e8c630
commit 0e87e0c7de
8 changed files with 69 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
kittens/clipboard/legacy.go
View File

@@ -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()
}

4
kittens/clipboard/write.go
View File

@@ -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()
}

2
kittens/transfer/receive.go
View File

@@ -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 {

116
kittens/transfer/send.go
View File

@@ -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
}

4
tools/cmd/at/tty_io.go
View File

@@ -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)

2
tools/tui/loop/api.go
View File

@@ -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

3
tools/tui/loop/run.go
View File

@@ -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
}

8
tools/tui/loop/write.go
View File

@@ -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")
}