Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

added missing insert fails counter #383

Merged
merged 4 commits into from
Mar 17, 2015
Merged

added missing insert fails counter #383

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8c7d2d4
added missing insert fails counter
MansurAshraf Dec 16, 2014
16e034a
added missing counters
MansurAshraf Dec 17, 2014
a8f53a9
renamed put counter
MansurAshraf Dec 18, 2014
f76a781
Merge remote-tracking branch 'upstream/develop' into feature/insertFail
MansurAshraf Mar 17, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions algebird-util/src/main/scala/com/twitter/algebird/util/summer/AsyncListSum.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@ class AsyncListSum[Key, Value](bufferSize: BufferSize,

@annotation.tailrec
private[this] final def doInsert(key: Key, value: Value) {
tuplesIn.incr
val (success, countChange) = if (queueMap.containsKey(key)) {
val oldValue = queueMap.get(key)
if (oldValue != null) {
Expand All @@ -120,6 +121,7 @@ class AsyncListSum[Key, Value](bufferSize: BufferSize,
// Successful insert
elementsInCache.addAndGet(countChange)
} else {
insertFails.incr
return doInsert(key, value)
}
}
Expand Down
4 changes: 2 additions & 2 deletions algebird-util/src/main/scala/com/twitter/algebird/util/summer/SyncSummingQueue.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,14 +33,14 @@ case class SyncSummingQueue[Key, Value](bufferSize: BufferSize,
override val memoryIncr: Incrementor,
override val timeoutIncr: Incrementor,
sizeIncr: Incrementor,
putCalls: Incrementor,
insertOps: Incrementor,
tuplesIn: Incrementor,
tuplesOut: Incrementor)(implicit semigroup: Semigroup[Value]) extends AsyncSummer[(Key, Value), Map[Key, Value]] with WithFlushConditions[(Key, Value), Map[Key, Value]] {

require(bufferSize.v > 0, "Use the Null summer for an empty async summer")
protected override val emptyResult = Map.empty[Key, Value]

private[this] final val squeue: CustomSummingQueue[Map[Key, Value]] = new CustomSummingQueue(bufferSize.v, sizeIncr, putCalls)
private[this] final val squeue: CustomSummingQueue[Map[Key, Value]] = new CustomSummingQueue(bufferSize.v, sizeIncr, insertOps)
override def isFlushed: Boolean = squeue.isFlushed

def flush: Future[Map[Key, Value]] = {
Expand Down