WIP: Added AIPerService::Approvement

indra/llmessage/aicurlperservice.cpp

@@ -71,7 +71,7 @@ void intrusive_ptr_release(RefCountedThreadSafePerService* per_service)
 using namespace AICurlPrivate;
 
 AIPerService::AIPerService(void) :
-		mQueuedCommands(0), mAdded(0), mQueueEmpty(false),
+		mApprovedRequests(0), mQueuedCommands(0), mAdded(0), mQueueEmpty(false),
 		mQueueFull(false), mRequestStarvation(false), mHTTPBandwidth(25),	// 25 = 1000 ms / 40 ms.
 		mConcurrectConnections(CurlConcurrentConnectionsPerService),
 		mMaxPipelinedRequests(CurlConcurrentConnectionsPerService)
@@ -355,3 +355,14 @@ void AIPerService::adjust_concurrent_connections(int increment)
   }
 }
 
+void AIPerService::Approvement::honored(void)
+{
+  if (!mHonored)
+  {
+	mHonored = true;
+	AICurlPrivate::PerService_wat per_service_w(*mPerServicePtr);
+	llassert(per_service_w->mApprovedRequests > 0);
+	per_service_w->mApprovedRequests--;
+  }
+}
+

indra/llmessage/aicurlperservice.h

@@ -114,6 +114,7 @@ class AIPerService {
   private:
 	typedef std::deque<AICurlPrivate::BufferedCurlEasyRequestPtr> queued_request_type;
 
+	int mApprovedRequests;						// The number of approved requests by wantsMoreHTTPRequestsFor that were not added to the command queue yet.
 	int mQueuedCommands;						// Number of add commands (minus remove commands) with this host in the command queue.
 	int mAdded;									// Number of active easy handles with this host.
 	queued_request_type mQueuedRequests;		// Waiting (throttled) requests.
@@ -188,7 +189,7 @@ class AIPerService {
 														// Add queued easy handle (if any) to the multi handle. The request is removed from the queue,
 														// followed by either a call to added_to_multi_handle() or to queue() to add it back.
 
-	S32 pipelined_requests(void) const { return mQueuedCommands + mQueuedRequests.size() + mAdded; }
+	S32 pipelined_requests(void) const { return mApprovedRequests + mQueuedCommands + mQueuedRequests.size() + mAdded; }
 
 	AIAverage& bandwidth(void) { return mHTTPBandwidth; }
 	AIAverage const& bandwidth(void) const { return mHTTPBandwidth; }
@@ -212,6 +213,17 @@ class AIPerService {
 	// Return true if too much bandwidth is being used.
 	static bool checkBandwidthUsage(AIPerServicePtr const& per_service, U64 sTime_40ms);
 
+	// A helper class to decrement mApprovedRequests after requests approved by wantsMoreHTTPRequestsFor were handled.
+	class Approvement {
+	  private:
+		AIPerServicePtr mPerServicePtr;
+		bool mHonored;
+	  public:
+		Approvement(AIPerServicePtr const& per_service) : mPerServicePtr(per_service), mHonored(false) { }
+		~Approvement() { honored(); }
+		void honored(void);
+	};
+
   private:
 	// Disallow copying.
 	AIPerService(AIPerService const&);

indra/llmessage/aicurlthread.cpp

@@ -2650,6 +2650,7 @@ bool AIPerService::wantsMoreHTTPRequestsFor(AIPerServicePtr const& per_service)
   {
 	PerService_wat per_service_w(*per_service);
 	S32 const pipelined_requests_per_service = per_service_w->pipelined_requests();
+	llassert(pipelined_requests_per_service >= 0 && pipelined_requests_per_service <= 16);
 	reject = pipelined_requests_per_service >= per_service_w->mMaxPipelinedRequests;
 	equal = pipelined_requests_per_service == per_service_w->mMaxPipelinedRequests;
 	increment_threshold = per_service_w->mRequestStarvation;
@@ -2672,6 +2673,13 @@ bool AIPerService::wantsMoreHTTPRequestsFor(AIPerServicePtr const& per_service)
 		reject = !equal;
 	  }
 	}
+	if (!reject)
+	{
+	  // Before releasing the lock on per_service, stop other threads from getting a
+	  // too small value from pipelined_requests() and approving too many requests.
+	  per_service_w->mApprovedRequests++;
+	  llassert(per_service_w->mApprovedRequests <= 16);
+	}
   }
   if (reject)
   {
@@ -2683,6 +2691,7 @@ bool AIPerService::wantsMoreHTTPRequestsFor(AIPerServicePtr const& per_service)
   if (checkBandwidthUsage(per_service, sTime_40ms))
   {
 	// Too much bandwidth is being used, either in total or for this service.
+	PerService_wat(*per_service)->mApprovedRequests--;		// Not approved after all.
 	return false;
   }
 
@@ -2714,6 +2723,10 @@ bool AIPerService::wantsMoreHTTPRequestsFor(AIPerServicePtr const& per_service)
 	  reject = !equal;
 	}
   }
+  if (reject)
+  {
+	PerService_wat(*per_service)->mApprovedRequests--;		// Not approved after all.
+  }
   return !reject;
 }
 
@@ -2773,6 +2786,8 @@ bool AIPerService::checkBandwidthUsage(AIPerServicePtr const& per_service, U64 s
   }
 
   // Throttle this service if it uses too much bandwidth.
+  // Warning: this requires max_bandwidth * 1024 to fit in a size_t.
+  // On 32 bit that means that HTTPThrottleBandwidth must be less than 33554 kbps.
   return (service_bandwidth > (max_bandwidth * throttle_fraction_w->fraction / 1024));
 }