AU: Fix potential issues with premature destruction of HTTP fetchers.

multi_http_fetcher.h

 // Copyright (c) 2010 The Chromium OS Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #ifndef CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__
 #define CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__
 
 #include <tr1/memory>
 #include <utility>
 #include <vector>
 
 #include "update_engine/http_fetcher.h"
+#include "update_engine/utils.h"
 
 // This class is a simple wrapper around an HttpFetcher. The client
 // specifies a vector of byte ranges. MultiHttpFetcher will fetch bytes
 // from those offsets. Pass -1 as a length to specify unlimited length.
 // It really only would make sense for the last range specified to have
 // unlimited length.
 
 namespace chromeos_update_engine {
 
 template<typename BaseHttpFetcher>
 class MultiHttpFetcher : public HttpFetcher, public HttpFetcherDelegate {
  public:
   typedef std::vector<std::pair<off_t, off_t> > RangesVect;
 
   MultiHttpFetcher()
       : sent_transfer_complete_(false),
+        pending_next_fetcher_(false),
         current_index_(0),
         bytes_received_this_fetcher_(0) {}
   ~MultiHttpFetcher() {}
 
   void set_ranges(const RangesVect& ranges) {
     ranges_ = ranges;
     fetchers_.resize(ranges_.size());  // Allocate the fetchers
     for (typename std::vector<std::tr1::shared_ptr<BaseHttpFetcher>
              >::iterator it = fetchers_.begin(), e = fetchers_.end();
          it != e; ++it) {
@@ skipping 10 matching lines @@
     if (ranges_.empty()) {
       if (delegate_)
         delegate_->TransferComplete(this, true);
       return;
     }
     current_index_ = 0;
     LOG(INFO) << "starting first transfer";
     StartTransfer();
   }
 
-  void TerminateTransfer() {
-    if (current_index_ < fetchers_.size())
-      fetchers_[current_index_]->TerminateTransfer();
+  void TransferTerminated(HttpFetcher* fetcher) {
+    LOG(INFO) << "Received transfer terminated.";
+    if (pending_next_fetcher_) {
+      pending_next_fetcher_ = false;
+      if (++current_index_ >= ranges_.size()) {
+        SendTransferComplete(fetcher, true);
+      } else {
+        StartTransfer();
+      }
+      return;
+    }
     current_index_ = ranges_.size();
     sent_transfer_complete_ = true;  // a fib
+    if (delegate_) {
+      // Note that after the callback returns this object may be destroyed.
+      delegate_->TransferTerminated(this);
+    }
+  }
+
+  void TerminateTransfer() {
+    // If the current fetcher is already being terminated, just wait for its
+    // TransferTerminated callback.
+    if (pending_next_fetcher_) {
+      pending_next_fetcher_ = false;
+      return;
+    }
+    // If there's a current active fetcher terminate it and wait for its
+    // TransferTerminated callback.
+    if (current_index_ < fetchers_.size()) {
+      fetchers_[current_index_]->TerminateTransfer();
+      return;
+    }
+    TransferTerminated(this);

[adlr, 2010/11/17 21:33:50]

i'm confused as to what sequence of events would lead this line to get executed.
can you tell me?

thanks

[petkov, 2010/11/17 21:47:54]

I added a comment in the code. I guess the only case is when TerminateTransfer
is invoked before set_ranges.

   }
 
   void Pause() {
     if (current_index_ < fetchers_.size())
       fetchers_[current_index_]->Pause();
   }
 
   void Unpause() {
     if (current_index_ < fetchers_.size())
       fetchers_[current_index_]->Unpause();
@@ skipping 46 matching lines @@
     }
     LOG(INFO) << "Starting a transfer @" << ranges_[current_index_].first << "("
               << ranges_[current_index_].second << ")";
     bytes_received_this_fetcher_ = 0;
     fetchers_[current_index_]->SetOffset(ranges_[current_index_].first);
     if (delegate_)
       delegate_->SeekToOffset(ranges_[current_index_].first);
     fetchers_[current_index_]->BeginTransfer(url_);
   }
 
-  void ReceivedBytes(HttpFetcher* fetcher,
-                     const char* bytes,
-                     int length) {
-    if (current_index_ >= ranges_.size())
-      return;
-    if (fetcher != fetchers_[current_index_].get()) {
-      LOG(WARNING) << "Received bytes from invalid fetcher";
-      return;
-    }
+  void ReceivedBytes(HttpFetcher* fetcher, const char* bytes, int length) {
+    TEST_AND_RETURN(current_index_ < ranges_.size());
+    TEST_AND_RETURN(fetcher == fetchers_[current_index_].get());
+    TEST_AND_RETURN(!pending_next_fetcher_);
     off_t next_size = length;
     if (ranges_[current_index_].second >= 0) {
       next_size = std::min(next_size,
                            ranges_[current_index_].second -
                            bytes_received_this_fetcher_);
     }
     LOG_IF(WARNING, next_size <= 0) << "Asked to write length <= 0";
     if (delegate_) {
       delegate_->ReceivedBytes(this, bytes, next_size);
     }
     bytes_received_this_fetcher_ += length;
     if (ranges_[current_index_].second >= 0 &&
         bytes_received_this_fetcher_ >= ranges_[current_index_].second) {
-      fetchers_[current_index_]->TerminateTransfer();
-      current_index_++;
-      if (current_index_ == ranges_.size()) {
-        SendTransferComplete(fetchers_[current_index_ - 1].get(), true);
-      } else {
-        StartTransfer();
-      }
+      // Terminates the current fetcher. Waits for its TransferTerminated
+      // callback before starting the next fetcher so that we don't end up
+      // signalling the delegate that the whole multi-transfer is complete
+      // before all fetchers are really done and cleaned up.
+      pending_next_fetcher_ = true;
+      fetcher->TerminateTransfer();
     }
   }
 
   void TransferComplete(HttpFetcher* fetcher, bool successful) {
-    LOG(INFO) << "Received transfer complete";
+    TEST_AND_RETURN(!pending_next_fetcher_);
+    LOG(INFO) << "Received transfer complete.";
     if (current_index_ >= ranges_.size()) {
       SendTransferComplete(fetcher, true);
       return;
     }
-
     if (ranges_[current_index_].second < 0) {
       // We're done with the current operation
       current_index_++;
       if (current_index_ >= ranges_.size() || !successful) {
         SendTransferComplete(fetcher, successful);
       } else {
         // Do the next transfer
         StartTransfer();
       }
       return;
     }
-
     if (bytes_received_this_fetcher_ < ranges_[current_index_].second) {
       LOG(WARNING) << "Received insufficient bytes from fetcher. "
                    << "Ending early";
       SendTransferComplete(fetcher, false);
       return;
-    } else {
-      LOG(INFO) << "Got spurious TransferComplete. Ingoring.";
     }
+    LOG(INFO) << "Got spurious TransferComplete. Ignoring.";
   }
 
   // If true, do not send any more data or TransferComplete to the delegate.
   bool sent_transfer_complete_;
 
+  // If true, the next fetcher needs to be started when TransferTerminated is
+  // received from the current fetcher.
+  bool pending_next_fetcher_;
+
   RangesVect ranges_;
   std::vector<std::tr1::shared_ptr<BaseHttpFetcher> > fetchers_;
 
   RangesVect::size_type current_index_;  // index into ranges_, fetchers_
   off_t bytes_received_this_fetcher_;
 
- private:
   DISALLOW_COPY_AND_ASSIGN(MultiHttpFetcher);
 };
 
 }  // namespace chromeos_update_engine
 
 #endif  // CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__