-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
optionally enable concurrency in action containers
- Loading branch information
1 parent
33bb0e7
commit 789e7a3
Showing
57 changed files
with
2,041 additions
and
238 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
116 changes: 116 additions & 0 deletions
116
common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.openwhisk.common | ||
|
|
||
| import scala.collection.concurrent.TrieMap | ||
|
|
||
| /** | ||
| * A Semaphore that coordinates the memory (ForcibleSemaphore) and concurrency (ResizableSemaphore) where | ||
| * - for invocations when maxConcurrent == 1, delegate to super | ||
| * - for invocations that cause acquire on memory slots, also acquire concurrency slots, and do it atomically | ||
| * @param memoryPermits | ||
| * @tparam T | ||
| */ | ||
| class NestedSemaphore[T](memoryPermits: Int) extends ForcibleSemaphore(memoryPermits) { | ||
| private val actionConcurrentSlotsMap = TrieMap.empty[T, ResizableSemaphore] //one key per action; resized per container | ||
|
|
||
| final def tryAcquireConcurrent(actionid: T, maxConcurrent: Int, memoryPermits: Int): Boolean = { | ||
|
|
||
| if (maxConcurrent == 1) { | ||
| super.tryAcquire(memoryPermits) | ||
| } else { | ||
| tryOrForceAcquireConcurrent(actionid, maxConcurrent, memoryPermits, false) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Coordinated permit acquisition: | ||
| * - first try to acquire concurrency slot | ||
| * - then try to acquire lock for this action | ||
| * - within the lock: | ||
| * - try to acquire concurrency slot (double check) | ||
| * - try to acquire memory slot | ||
| * - if memory slot acquired, release concurrency slots | ||
| * - release the lock | ||
| * - if neither concurrency slot nor memory slot acquired, return false | ||
| * @param actionid | ||
| * @param maxConcurrent | ||
| * @param memoryPermits | ||
| * @param force | ||
| * @return | ||
| */ | ||
| private def tryOrForceAcquireConcurrent(actionid: T, | ||
| maxConcurrent: Int, | ||
| memoryPermits: Int, | ||
| force: Boolean): Boolean = { | ||
| val concurrentSlots = actionConcurrentSlotsMap | ||
| .getOrElseUpdate(actionid, new ResizableSemaphore(0, maxConcurrent)) | ||
| if (concurrentSlots.tryAcquire(1)) { | ||
| true | ||
| } else { | ||
| // with synchronized: | ||
| concurrentSlots.synchronized { | ||
| if (concurrentSlots.tryAcquire(1)) { | ||
| true | ||
| } else if (force) { | ||
| super.forceAcquire(memoryPermits) | ||
| concurrentSlots.release(maxConcurrent - 1, false) | ||
| true | ||
| } else if (super.tryAcquire(memoryPermits)) { | ||
| concurrentSlots.release(maxConcurrent - 1, false) | ||
| true | ||
| } else { | ||
| false | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def forceAcquireConcurrent(actionid: T, maxConcurrent: Int, memoryPermits: Int): Unit = { | ||
| require(memoryPermits > 0, "cannot force acquire negative or no permits") | ||
| if (maxConcurrent == 1) { | ||
| super.forceAcquire(memoryPermits) | ||
| } else { | ||
| tryOrForceAcquireConcurrent(actionid, maxConcurrent, memoryPermits, true) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Releases the given amount of permits | ||
| * | ||
| * @param acquires the number of permits to release | ||
| */ | ||
| def releaseConcurrent(actionid: T, maxConcurrent: Int, memoryPermits: Int): Unit = { | ||
| require(memoryPermits > 0, "cannot release negative or no permits") | ||
| if (maxConcurrent == 1) { | ||
| super.release(memoryPermits) | ||
| } else { | ||
| val concurrentSlots = actionConcurrentSlotsMap(actionid) | ||
| val (memoryRelease, actionRelease) = concurrentSlots.release(1, true) | ||
| //concurrent slots | ||
| if (memoryRelease) { | ||
| super.release(memoryPermits) | ||
| } | ||
| if (actionRelease) { | ||
| actionConcurrentSlotsMap.remove(actionid) | ||
| } | ||
| } | ||
| } | ||
| //for testing | ||
| def concurrentState = actionConcurrentSlotsMap.readOnlySnapshot() | ||
| } |
115 changes: 115 additions & 0 deletions
115
common/scala/src/main/scala/org/apache/openwhisk/common/ResizableSemaphore.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.openwhisk.common | ||
|
|
||
| import java.util.concurrent.atomic.AtomicInteger | ||
| import java.util.concurrent.locks.AbstractQueuedSynchronizer | ||
| import scala.annotation.tailrec | ||
|
|
||
| /** | ||
| * A Semaphore that has a specialized release process that optionally allows reduction of permits in batches. | ||
| * When permit size after release is a factor of reductionSize, the release process will reset permits to state + 1 - reductionSize; | ||
| * otherwise the release will reset permits to state + 1. | ||
| * It also maintains an operationCount where a tryAquire + release is a single operation, | ||
| * so that we can know once all operations are completed. | ||
| * @param maxAllowed | ||
| * @param reductionSize | ||
| */ | ||
| class ResizableSemaphore(maxAllowed: Int, reductionSize: Int) { | ||
| private val operationCount = new AtomicInteger(0) | ||
| class Sync extends AbstractQueuedSynchronizer { | ||
| setState(maxAllowed) | ||
|
|
||
| def permits: Int = getState | ||
|
|
||
| /** Try to release a permit and return whether or not that operation was successful. */ | ||
| @tailrec | ||
| final def tryReleaseSharedWithResult(releases: Int): Boolean = { | ||
| val current = getState | ||
| val next2 = current + releases | ||
| val (next, reduced) = if (next2 % reductionSize == 0) { | ||
| (next2 - reductionSize, true) | ||
| } else { | ||
| (next2, false) | ||
| } | ||
| //next MIGHT be < current in case of reduction; this is OK!!! | ||
| if (compareAndSetState(current, next)) { | ||
| reduced | ||
| } else { | ||
| tryReleaseSharedWithResult(releases) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Try to acquire a permit and return whether or not that operation was successful. Requests may not finish in FIFO | ||
| * order, hence this method is not necessarily fair. | ||
| */ | ||
| @tailrec | ||
| final def nonFairTryAcquireShared(acquires: Int): Int = { | ||
| val available = getState | ||
| val remaining = available - acquires | ||
| if (remaining < 0 || compareAndSetState(available, remaining)) { | ||
| remaining | ||
| } else { | ||
| nonFairTryAcquireShared(acquires) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| val sync = new Sync | ||
|
|
||
| /** | ||
| * Acquires the given numbers of permits. | ||
| * | ||
| * @param acquires the number of permits to get | ||
| * @return `true`, iff the internal semaphore's number of permits is positive, `false` if negative | ||
| */ | ||
| def tryAcquire(acquires: Int = 1): Boolean = { | ||
| require(acquires > 0, "cannot acquire negative or no permits") | ||
| if (sync.nonFairTryAcquireShared(acquires) >= 0) { | ||
| operationCount.incrementAndGet() | ||
| true | ||
| } else { | ||
| false | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Releases the given amount of permits | ||
| * | ||
| * @param acquires the number of permits to release | ||
| * @return (releaseMemory, releaseAction) releaseMemory is true if concurrency count is a factor of reductionSize | ||
| * releaseAction is true if the operationCount reaches 0 | ||
| */ | ||
| def release(acquires: Int = 1, opComplete: Boolean): (Boolean, Boolean) = { | ||
| require(acquires > 0, "cannot release negative or no permits") | ||
| //release always succeeds, so we can always adjust the operationCount | ||
| val releaseAction = if (opComplete) { // an operation completion | ||
| operationCount.decrementAndGet() == 0 | ||
| } else { //otherwise an allocation + operation initialization | ||
| operationCount.incrementAndGet() == 0 | ||
| } | ||
| (sync.tryReleaseSharedWithResult(acquires), releaseAction) | ||
| } | ||
|
|
||
| /** Returns the number of currently available permits. Possibly negative. */ | ||
| def availablePermits: Int = sync.permits | ||
|
|
||
| //for testing | ||
| def counter = operationCount.get() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.