DownloadFinished This function will be called when the node found new block are finalized, and it will update the local L1 view and commit new blobs into local storage file.
(newL1 int64, kvIndices []uint64, blobs [][]byte, commits []common.Hash)
| 107 | // DownloadFinished This function will be called when the node found new block are finalized, and it will update the |
| 108 | // local L1 view and commit new blobs into local storage file. |
| 109 | func (s *StorageManager) DownloadFinished(newL1 int64, kvIndices []uint64, blobs [][]byte, commits []common.Hash) error { |
| 110 | if len(kvIndices) != len(blobs) || len(blobs) != len(commits) { |
| 111 | return errors.New("invalid params lens") |
| 112 | } |
| 113 | |
| 114 | s.mu.Lock() |
| 115 | defer s.mu.Unlock() |
| 116 | |
| 117 | // in most case, newL1 should be equal to s.localL1 + 32 |
| 118 | // but it is possible that the node was shutdown for some time, and when it restart and DownloadFinished for the first time |
| 119 | // the new finalized L1 will be larger than that, so we just do the simple compare check here. |
| 120 | if newL1 <= s.localL1 { |
| 121 | return errors.New("new L1 is older than local L1") |
| 122 | } |
| 123 | |
| 124 | taskNum := s.DownloadThreadNum |
| 125 | var wg sync.WaitGroup |
| 126 | chanRes := make(chan error, taskNum) |
| 127 | defer close(chanRes) |
| 128 | |
| 129 | taskIdx := 0 |
| 130 | for taskIdx < taskNum { |
| 131 | if taskIdx >= len(kvIndices) { |
| 132 | break |
| 133 | } |
| 134 | |
| 135 | wg.Add(1) |
| 136 | |
| 137 | insertIdxInTask := make([]int, 0) |
| 138 | for i := taskIdx; i < len(kvIndices); i += taskNum { |
| 139 | insertIdxInTask = append(insertIdxInTask, i) |
| 140 | } |
| 141 | |
| 142 | go func(insertIdx []int, out chan<- error) { |
| 143 | defer wg.Done() |
| 144 | |
| 145 | var err error = nil |
| 146 | for _, idx := range insertIdx { |
| 147 | c := PrepareCommit(commits[idx]) |
| 148 | done, err := s.shardManager.TryWriteEncoded(kvIndices[idx], blobs[idx], c) |
| 149 | if err != nil { |
| 150 | break |
| 151 | } |
| 152 | if !done { |
| 153 | shardIdx := kvIndices[idx] / s.KvEntries() |
| 154 | s.lg.Warn("Shard not managed locally to store the blob", "kvIndex", kvIndices[idx], "shardIdx", shardIdx) |
| 155 | } |
| 156 | } |
| 157 | |
| 158 | chanRes <- err |
| 159 | }(insertIdxInTask, chanRes) |
| 160 | |
| 161 | taskIdx++ |
| 162 | } |
| 163 | |
| 164 | wg.Wait() |
| 165 | |
| 166 | for i := 0; i < taskIdx; i++ { |