Additional bug-fixes found looking at PR-361

This commit is contained in:
Cassondra Foesch 2020-07-18 19:45:27 +00:00
parent 97b9df616e
commit e56b4ff6ad
2 changed files with 70 additions and 43 deletions

View File

@ -106,6 +106,35 @@ func (rs *RequestServer) closeRequest(handle string) error {
// Close the read/write/closer to trigger exiting the main server loop
func (rs *RequestServer) Close() error { return rs.conn.Close() }
func (rs *RequestServer) serveLoop(pktChan chan<- orderedRequest) error {
var err error
var pkt requestPacket
var pktType uint8
var pktBytes []byte
for {
pktType, pktBytes, err = rs.serverConn.recvPacket(rs.pktMgr.getNextOrderID())
if err != nil {
// we don't care about releasing allocated pages here, the server will quit and the allocator freed
return err
}
pkt, err = makePacket(rxPacket{fxp(pktType), pktBytes})
if err != nil {
switch errors.Cause(err) {
case errUnknownExtendedPacket:
// do nothing
default:
debug("makePacket err: %v", err)
rs.conn.Close() // shuts down recvPacket
return err
}
}
pktChan <- rs.pktMgr.newOrderedRequest(pkt)
}
}
// Serve requests for user session
func (rs *RequestServer) Serve() error {
defer func() {
@ -127,51 +156,19 @@ func (rs *RequestServer) Serve() error {
}
pktChan := rs.pktMgr.workerChan(runWorker)
var err error
var pkt requestPacket
var pktType uint8
var pktBytes []byte
for {
pktType, pktBytes, err = rs.serverConn.recvPacket(rs.pktMgr.getNextOrderID())
if err != nil {
// we don't care about releasing allocated pages here, the server will quit and the allocator freed
break
}
pkt, err = makePacket(rxPacket{fxp(pktType), pktBytes})
if err != nil {
switch errors.Cause(err) {
case errUnknownExtendedPacket:
// do nothing
default:
debug("makePacket err: %v", err)
rs.conn.Close() // shuts down recvPacket
break
}
}
pktChan <- rs.pktMgr.newOrderedRequest(pkt)
}
err := rs.serveLoop(pktChan)
close(pktChan) // shuts down sftpServerWorkers
wg.Wait() // wait for all workers to exit
rs.openRequestLock.Lock()
defer rs.openRequestLock.Unlock()
// make sure all open requests are properly closed
// (eg. possible on dropped connections, client crashes, etc.)
for handle, req := range rs.openRequests {
if err != nil {
req.state.RLock()
writer := req.state.writerAt
reader := req.state.readerAt
req.state.RUnlock()
if t, ok := writer.(TransferError); ok {
debug("notify error: %v to writer: %v\n", err, writer)
t.TransferError(err)
}
if t, ok := reader.(TransferError); ok {
debug("notify error: %v to reader: %v\n", err, reader)
t.TransferError(err)
}
}
req.transferError(err)
delete(rs.openRequests, handle)
req.close()
}

View File

@ -138,19 +138,49 @@ func (r *Request) close() error {
r.cancelCtx()
}
}()
r.state.RLock()
rd := r.state.readerAt
wr := r.state.writerAt
r.state.RUnlock()
var err error
if c, ok := rd.(io.Closer); ok {
return c.Close()
if err2 := c.Close(); err == nil {
// update error if it is still nil
err = err2
}
}
if c, ok := wr.(io.Closer); ok {
if err2 := c.Close(); err == nil {
// update error if it is still nil
err = err2
}
}
return err
}
// Close reader/writer if possible
func (r *Request) transferError(err error) {
if err == nil {
return
}
r.state.RLock()
wt := r.state.writerAt
rd := r.state.readerAt
wr := r.state.writerAt
r.state.RUnlock()
if c, ok := wt.(io.Closer); ok {
return c.Close()
if t, ok := rd.(TransferError); ok {
t.TransferError(err)
}
if t, ok := wr.(TransferError); ok {
t.TransferError(err)
}
return nil
}
// called from worker to handle packet/request