Use semaphore in ObjectPool #426

harmandeeps · 2020-07-13T08:15:51Z

The intent of the change is to remove the synchronization block while getting/returning objects from pool and use a semaphore to enable parallel object creation.

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

harmandeeps

Thank you for the review @stagraqubole.

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

rohangarg · 2020-07-23T13:47:29Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

-      decreaseObject(object);
+      log.debug(String.format("Returning object: %s to queue. Queue size: %d", object, objectQueue.size()));
+      if (!objectQueue.offer(object)) {
+        log.warn("Created more objects than configured. Created=" + totalCount + " QueueSize=" + objectQueue.size());


is it ok to release the semaphore incase more objects were created than maxSize

No, this needs a change

maybe I am missing something. Theoretically, this case should happen as we are controlling parallelism via counter-based semaphore.
suppose, this happens, then why shouldn't we release the semaphore as the returning thread is done with the object? won't it give a chance to other waiting threads?

if we are controlling parallelism strictly via counter-based semaphore, then this is an error state. We should error out if this happens instead.
I don't think it makes sense to operate a semaphore if the number of objects can be larger than permits (it can lead to wrong calculations).

yes, true. throwing error in such a case.

rohangarg · 2020-07-23T13:49:43Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

    try {
      if (blocking) {
-        freeObject = objectQueue.take();
+        takeSemaphore.acquire();
+        object = objectQueue.take();


what's the reason for not calling tryGetObject here?

We should. Or we should remove the code path for blocking=true case as it is not applicable for us and it is only complicating maintenance.

tryGetObject call object.Queue.poll() which is non blocking. But, in case of blocking mode, though there is semaphore acquire blocking call, I just a called 'take' to be on safe side.

I guess better removing this blocking mode as we are not using it anymore in the code.

rohangarg · 2020-07-24T12:14:46Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

  private final String host;
  private final int socketTimeout;
  private final int connectTimeout;
+  private final Semaphore takeSemaphore;
+  private final AtomicInteger totalCount;


is it possible to remove totalCount and rather use takeSemaphore#permits

yeah, will change in next iteration.

I checked it is not possible. We initialize the pool with min size without acquiring semaphore.

rohangarg · 2020-07-24T12:16:54Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

  }

-  public synchronized int getTotalCount()
+  public int getTotalCount()


should this be getActiveObjectCount (the var too)?

will change this.

rohangarg · 2020-07-24T12:28:34Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

      }
+      object.setLastAccessTs(System.currentTimeMillis());


why is this set while giving the object? what if this object is being used for more than the max-idle time?
shouldn't this be set to some sentinel value while giving the object, and to a timestamp while release of the object of the pool?

We should fix this separately, I will open a issue. Right now it works because thrift call time is much less than idle timeout.

What do you mean by max-idle time? If you are referring scavenging the under connection, then in this case , object won't be present in the queue to get scavenged.

All the objects are inherently same (don't have any state properties associated), setting timestamp while getting or returning shouldn't matter much. Or am I missing something?

Agree with Shubham as to why it works for now.

setting timestamp while getting or returning shouldn't matter much.

Ok, if this will not be a part of scavenger thread, then doesn't make sense to set the last access time in release method? borrow is logically not the last access time according to me.

yeah, got it. makes sense. for example, if in a query burst all objects are taken for 20 minutes (suppose) and the released. Scanvenger will think these objects are not getting used and scavenge.

We can take this in a separate commit as subhuam mentioned.

rohangarg · 2020-07-28T12:01:03Z

Also, can you please the exact motivation for using semaphore in the object pool?
Were things like atomic counter + mutex locking or blocking queue not enough for controlling concurrency? (and why if they were not).

harmandeeps · 2020-09-30T10:28:42Z

Also, can you please the exact motivation for using semaphore in the object pool?
Were things like atomic counter + mutex locking or blocking queue not enough for controlling concurrency? (and why if they were not).

changed the description. please check.

shubhamtagra · 2020-10-01T11:58:29Z

rubix-spi/src/main/java/com/qubole/rubix/spi/CacheConfig.java

@@ -310,7 +308,7 @@ public static int getTransportPoolMaxWait(Configuration conf)

  public static int getScavengeInterval(Configuration conf)
  {
-    return conf.getInt(KEY_POOL_MAX_WAIT_TIMEOUT, DEFAULT_SCAVENGE_INTERVAL);
+    return conf.getInt(KEY_POOL_SCAVENGER_INTERVAL, DEFAULT_SCAVENGE_INTERVAL);


please separate out commits. Have one separate for changing config name

created a separate commit.

shubhamtagra · 2020-10-01T12:32:05Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java

+  {
+    Poolable<T> object;
+    try {
+      takeSemaphore.tryAcquire(config.getMaxWaitMilliseconds(), TimeUnit.MILLISECONDS);


Return value denotes timeout. If we timeout then we need to return instead of going ahead without permit, that can cause creating more than max connections for some time. At the time of returning we would destroy extra ones and then double decrement semaphore.
Please add UTs around this to capture different cases.

Added logic to check the return type, if it is false (lock not acquired), return null. Also, added UT around the semaphore logic. Please have a look.

shubhamtagra · 2020-10-01T12:32:52Z

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPool.java

@@ -78,8 +76,11 @@ public void registerHost(String host)
    }
    log.debug(this.name + " : Borrowing object for partition: " + host);
    for (int i = 0; i < 3; i++) { // try at most three times


Remove this retry, better exit fast and fallback to direct reads.

…ect method

harmandeeps requested a review from shubhamtagra July 13, 2020 08:15

shubhamtagra requested changes Jul 17, 2020

View reviewed changes

harmandeeps commented Jul 23, 2020

View reviewed changes

harmandeeps force-pushed the adding-semaphore branch from 568bfda to bf78376 Compare July 23, 2020 12:49

rohangarg reviewed Jul 23, 2020

View reviewed changes

rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java Show resolved Hide resolved

rohangarg reviewed Jul 23, 2020

View reviewed changes

rohangarg reviewed Jul 24, 2020

View reviewed changes

harmandeeps force-pushed the adding-semaphore branch from 6f3d482 to 129a304 Compare September 30, 2020 10:19

harmandeeps force-pushed the adding-semaphore branch from 129a304 to a6cf9aa Compare October 1, 2020 12:35

shubhamtagra requested changes Oct 1, 2020

View reviewed changes

Harmandeep Singh added 4 commits October 6, 2020 15:38

Createing new config for connection pool scavanger interval

8e1c8e1

Removed synchronized block from increaseObjects, refactored borrowObj…

bf60c93

…ect method

Use semaphore to allow parallel creation of pool connections.

8d976db

Added unit test around semaphore logic

1721dc0

harmandeeps force-pushed the adding-semaphore branch from e59dc21 to 1721dc0 Compare October 6, 2020 10:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use semaphore in ObjectPool #426

Use semaphore in ObjectPool #426

harmandeeps commented Jul 13, 2020 •

edited

Loading

harmandeeps left a comment

rohangarg Jul 23, 2020

shubhamtagra Jul 27, 2020

harmandeeps Jul 28, 2020

rohangarg Jul 28, 2020

harmandeeps Sep 30, 2020

rohangarg Jul 23, 2020

shubhamtagra Jul 27, 2020

harmandeeps Jul 28, 2020

rohangarg Jul 24, 2020

harmandeeps Jul 28, 2020

harmandeeps Jul 28, 2020

rohangarg Jul 24, 2020 •

edited

Loading

harmandeeps Jul 28, 2020

rohangarg Jul 24, 2020

shubhamtagra Jul 27, 2020

harmandeeps Jul 28, 2020

rohangarg Jul 28, 2020

harmandeeps Jul 28, 2020

rohangarg commented Jul 28, 2020

harmandeeps commented Sep 30, 2020

shubhamtagra Oct 1, 2020

harmandeeps Oct 6, 2020

shubhamtagra Oct 1, 2020

harmandeeps Oct 6, 2020

shubhamtagra Oct 1, 2020

harmandeeps Oct 6, 2020

Use semaphore in ObjectPool #426

Are you sure you want to change the base?

Use semaphore in ObjectPool #426

Conversation

harmandeeps commented Jul 13, 2020 • edited Loading

harmandeeps left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohangarg Jul 24, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohangarg commented Jul 28, 2020

harmandeeps commented Sep 30, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

harmandeeps commented Jul 13, 2020 •

edited

Loading

rohangarg Jul 24, 2020 •

edited

Loading