Keep downloading file opened

diff --git a/doc/news.ru.texi b/doc/news.ru.texi
index f4df62e64d9f8befb2172db3950d4984b6c9516b..bc42c6e54f5de99de0a7a631ac387277bb9a7182 100644 (file)
--- a/doc/news.ru.texi
+++ b/doc/news.ru.texi
@@ -11,6 +11,10 @@
 допустимого количества открытых файлов, если у вас было много пакетов в
 spool директории.
 
 допустимого количества открытых файлов, если у вас было много пакетов в
 spool директории.
 

+@item
+Оптимизация: не закрывать файловый дескриптор файла который мы качаем.
+Прежде каждый его кусочек приводил к дорогим open/close вызовам.
+

 @end itemize
 
 @node Релиз 6.0.0
 @end itemize
 
 @node Релиз 6.0.0

diff --git a/doc/news.texi b/doc/news.texi
index 68661dbb3c2b8ac3a322a2ef3c0a9ac68118160b..565f79e0af01e45fe54e3bdf6246792e1cdee7f6 100644 (file)
--- a/doc/news.texi
+++ b/doc/news.texi
@@ -12,6 +12,10 @@ Optimization: most commands do not keep opened file descriptors now.
 Previously you can exceed maximal number of opened files if you have got
 many packets in the spool directory.
 
 Previously you can exceed maximal number of opened files if you have got
 many packets in the spool directory.
 

+@item
+Optimization: do not close file descriptor of the file we download
+online. Previously each chunk lead to expensive open/close calls.
+

 @end itemize
 
 @node Release 6.0.0
 @end itemize
 
 @node Release 6.0.0

diff --git a/src/sp.go b/src/sp.go
index b5bad2d5dcd21a16b909c44a8ab29addf6d8a3a2..9410a2d905eb106a6cb2cbb53a7dca8b5895f4c5 100644 (file)
--- a/src/sp.go
+++ b/src/sp.go
@@ -59,6 +59,11 @@ var (
        spCheckerToken chan struct{}
 )
 
        spCheckerToken chan struct{}
 )
 

+type FdAndFullSize struct {
+       fd       *os.File
+       fullSize int64
+}
+

 type SPType uint8
 
 const (
 type SPType uint8
 
 const (


@@ -214,6 +219,7 @@ type SPState struct {
        listOnly       bool
        onlyPkts       map[[32]byte]bool
        writeSPBuf     bytes.Buffer
        listOnly       bool
        onlyPkts       map[[32]byte]bool
        writeSPBuf     bytes.Buffer

+       fds            map[string]FdAndFullSize

        sync.RWMutex
 }
 
        sync.RWMutex
 }
 


@@ -529,12 +535,21 @@ func (state *SPState) StartR(conn ConnDeadlined) error {
        return err
 }
 
        return err
 }
 

+func (state *SPState) closeFd(pth string) {
+       s, exists := state.fds[pth]
+       delete(state.fds, pth)
+       if exists {
+               s.fd.Close()
+       }
+}
+

 func (state *SPState) StartWorkers(
        conn ConnDeadlined,
        infosPayloads [][]byte,
        payload []byte,
 ) error {
        les := LEs{{"Node", state.Node.Id}, {"Nice", int(state.Nice)}}
 func (state *SPState) StartWorkers(
        conn ConnDeadlined,
        infosPayloads [][]byte,
        payload []byte,
 ) error {
        les := LEs{{"Node", state.Node.Id}, {"Nice", int(state.Nice)}}

+       state.fds = make(map[string]FdAndFullSize)

        state.isDead = make(chan struct{})
        if state.maxOnlineTime > 0 {
                state.mustFinishAt = state.started.Add(state.maxOnlineTime)
        state.isDead = make(chan struct{})
        if state.maxOnlineTime > 0 {
                state.mustFinishAt = state.started.Add(state.maxOnlineTime)


@@ -682,22 +697,29 @@ func (state *SPState) StartWorkers(
                                        {"Size", int64(freq.Offset)},
                                }...)
                                state.Ctx.LogD("sp-file", lesp, "queueing")
                                        {"Size", int64(freq.Offset)},
                                }...)
                                state.Ctx.LogD("sp-file", lesp, "queueing")

-                               fd, err := os.Open(filepath.Join(
+                               pth := filepath.Join(

                                        state.Ctx.Spool,
                                        state.Node.Id.String(),
                                        string(TTx),
                                        Base32Codec.EncodeToString(freq.Hash[:]),
                                        state.Ctx.Spool,
                                        state.Node.Id.String(),
                                        string(TTx),
                                        Base32Codec.EncodeToString(freq.Hash[:]),

-                               ))
-                               if err != nil {
-                                       state.Ctx.LogE("sp-file", lesp, err, "")
-                                       return
-                               }
-                               fi, err := fd.Stat()
-                               if err != nil {
-                                       state.Ctx.LogE("sp-file", lesp, err, "")
-                                       return
+                               )
+                               fdAndFullSize, exists := state.fds[pth]
+                               if !exists {
+                                       fd, err := os.Open(pth)
+                                       if err != nil {
+                                               state.Ctx.LogE("sp-file", lesp, err, "")
+                                               return
+                                       }
+                                       fi, err := fd.Stat()
+                                       if err != nil {
+                                               state.Ctx.LogE("sp-file", lesp, err, "")
+                                               return
+                                       }
+                                       fdAndFullSize = FdAndFullSize{fd: fd, fullSize: fi.Size()}
+                                       state.fds[pth] = fdAndFullSize

                                }
                                }

-                               fullSize := fi.Size()
+                               fd := fdAndFullSize.fd
+                               fullSize := fdAndFullSize.fullSize

                                var buf []byte
                                if freq.Offset < uint64(fullSize) {
                                        state.Ctx.LogD("sp-file", lesp, "seeking")
                                var buf []byte
                                if freq.Offset < uint64(fullSize) {
                                        state.Ctx.LogD("sp-file", lesp, "seeking")


@@ -714,7 +736,7 @@ func (state *SPState) StartWorkers(
                                        buf = buf[:n]
                                        state.Ctx.LogD("sp-file", append(lesp, LE{"Size", n}), "read")
                                }
                                        buf = buf[:n]
                                        state.Ctx.LogD("sp-file", append(lesp, LE{"Size", n}), "read")
                                }

-                               fd.Close() // #nosec G104
+                               state.closeFd(pth)

                                payload = MarshalSP(SPTypeFile, SPFile{
                                        Hash:    freq.Hash,
                                        Offset:  freq.Offset,
                                payload = MarshalSP(SPTypeFile, SPFile{
                                        Hash:    freq.Hash,
                                        Offset:  freq.Offset,


@@ -816,6 +838,9 @@ func (state *SPState) StartWorkers(
                state.wg.Done()
                state.SetDead()
                conn.Close() // #nosec G104
                state.wg.Done()
                state.SetDead()
                conn.Close() // #nosec G104

+               for _, s := range state.fds {
+                       s.fd.Close()
+               }

        }()
 
        return nil
        }()
 
        return nil


@@ -944,27 +969,34 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                string(TRx),
                        )
                        filePath := filepath.Join(dirToSync, Base32Codec.EncodeToString(file.Hash[:]))
                                string(TRx),
                        )
                        filePath := filepath.Join(dirToSync, Base32Codec.EncodeToString(file.Hash[:]))

+                       filePathPart := filePath + PartSuffix

                        state.Ctx.LogD("sp-file", lesp, "opening part")
                        state.Ctx.LogD("sp-file", lesp, "opening part")

-                       fd, err := os.OpenFile(
-                               filePath+PartSuffix,
-                               os.O_RDWR|os.O_CREATE,
-                               os.FileMode(0666),
-                       )
-                       if err != nil {
-                               state.Ctx.LogE("sp-file", lesp, err, "")
-                               return nil, err
+                       fdAndFullSize, exists := state.fds[filePathPart]
+                       var fd *os.File
+                       if exists {
+                               fd = fdAndFullSize.fd
+                       } else {
+                               fd, err = os.OpenFile(
+                                       filePathPart,
+                                       os.O_RDWR|os.O_CREATE,
+                                       os.FileMode(0666),
+                               )
+                               if err != nil {
+                                       state.Ctx.LogE("sp-file", lesp, err, "")
+                                       return nil, err
+                               }
+                               state.fds[filePathPart] = FdAndFullSize{fd: fd}

                        }
                        state.Ctx.LogD("sp-file", append(lesp, LE{"Offset", file.Offset}), "seeking")
                        if _, err = fd.Seek(int64(file.Offset), io.SeekStart); err != nil {
                                state.Ctx.LogE("sp-file", lesp, err, "")
                        }
                        state.Ctx.LogD("sp-file", append(lesp, LE{"Offset", file.Offset}), "seeking")
                        if _, err = fd.Seek(int64(file.Offset), io.SeekStart); err != nil {
                                state.Ctx.LogE("sp-file", lesp, err, "")

-                               fd.Close() // #nosec G104
+                               state.closeFd(filePathPart)

                                return nil, err
                        }
                        state.Ctx.LogD("sp-file", lesp, "writing")
                                return nil, err
                        }
                        state.Ctx.LogD("sp-file", lesp, "writing")

-                       _, err = fd.Write(file.Payload)
-                       if err != nil {
+                       if _, err = fd.Write(file.Payload); err != nil {

                                state.Ctx.LogE("sp-file", lesp, err, "")
                                state.Ctx.LogE("sp-file", lesp, err, "")

-                               fd.Close() // #nosec G104
+                               state.closeFd(filePathPart)

                                return nil, err
                        }
                        ourSize := int64(file.Offset + uint64(len(file.Payload)))
                                return nil, err
                        }
                        ourSize := int64(file.Offset + uint64(len(file.Payload)))


@@ -981,7 +1013,6 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                Progress("Rx", lesp)
                        }
                        if fullsize != ourSize {
                                Progress("Rx", lesp)
                        }
                        if fullsize != ourSize {

-                               fd.Close() // #nosec G104

                                continue
                        }
                        <-spCheckerToken
                                continue
                        }
                        <-spCheckerToken


@@ -991,19 +1022,19 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                }()
                                if err := fd.Sync(); err != nil {
                                        state.Ctx.LogE("sp-file", lesp, err, "sync")
                                }()
                                if err := fd.Sync(); err != nil {
                                        state.Ctx.LogE("sp-file", lesp, err, "sync")

-                                       fd.Close() // #nosec G104
+                                       state.closeFd(filePathPart)

                                        return
                                }
                                state.wg.Add(1)
                                defer state.wg.Done()
                                if _, err = fd.Seek(0, io.SeekStart); err != nil {
                                        return
                                }
                                state.wg.Add(1)
                                defer state.wg.Done()
                                if _, err = fd.Seek(0, io.SeekStart); err != nil {

-                                       fd.Close() // #nosec G104
+                                       state.closeFd(filePathPart)

                                        state.Ctx.LogE("sp-file", lesp, err, "")
                                        return
                                }
                                state.Ctx.LogD("sp-file", lesp, "checking")
                                gut, err := Check(fd, file.Hash[:], lesp, state.Ctx.ShowPrgrs)
                                        state.Ctx.LogE("sp-file", lesp, err, "")
                                        return
                                }
                                state.Ctx.LogD("sp-file", lesp, "checking")
                                gut, err := Check(fd, file.Hash[:], lesp, state.Ctx.ShowPrgrs)

-                               fd.Close() // #nosec G104
+                               state.closeFd(filePathPart)

                                if err != nil || !gut {
                                        state.Ctx.LogE("sp-file", lesp, errors.New("checksum mismatch"), "")
                                        return
                                if err != nil || !gut {
                                        state.Ctx.LogE("sp-file", lesp, errors.New("checksum mismatch"), "")
                                        return