修复bos断连后无限重试没有作用的问题 (PaddlePaddle#1315)

* 修复bos断连后无限重试没有作用的问题g

* 修复bos断连后无限重试没有作用的问题g

pkg/fs/client/cache/cache.go

@@ -92,6 +92,10 @@ func (r *rCache) readFromReadAhead(off int64, buf []byte) (bytesRead int, err er
 		nread, err = readAheadBuf.ReadAt(uint64(blockOff), buf[bytesRead:])
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			log.Errorf("readAheadBuf err %v nread %v", err.Error(), nread)
+			r.lock.Lock()
+			readAheadBuf.Buffer = nil
+			delete(r.buffers, indexOff)
+			r.lock.Unlock()
 			return 0, err
 		}
 		bytesRead += nread

pkg/fs/client/ufs/object/test_file.go

@@ -1672,5 +1672,27 @@ func a() int {
 	ac = 1
 	ac = 1
 	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
+	ac = 1
 	return ac
 }

pkg/fs/client/vfs/reader.go

@@ -116,6 +116,7 @@ func (fh *fileReader) Read(buf []byte, off uint64) (int, syscall.Errno) {
 					return 0, syscall.EBADF
 				}
 				if nread == 0 {
+					log.Infof("readFromStream nread 0 bytesRead %v off %v", bytesRead, off)
 					break
 				}
 			}
@@ -126,6 +127,7 @@ func (fh *fileReader) Read(buf []byte, off uint64) (int, syscall.Errno) {
 					return 0, syscall.EBADF
 				}
 				if nread == 0 {
+					log.Infof("ENOMEM readFromStream nread 0 bytesRead %v off %v", bytesRead, off)
 					break
 				}
 			}
@@ -170,8 +172,9 @@ func (fh *fileReader) readFromStream(off int64, buf []byte) (bytesRead int, err
 	}
 	if fh.streamReader == nil {
 		log.Debugf("init reader %s flags[%d] off[%d]", fh.path, fh.flags, off)
-		resp, err := fh.ufs.Get(fh.path, fh.flags, off, 0)
+		resp, err := fh.ufs.Get(fh.path, fh.flags, off, fh.size-off)
 		if err != nil {
+			log.Errorf("ufs get err %v", err)
 			return 0, err
 		}
 		fh.streamReader = resp