Return-Path: X-Original-To: apmail-drill-commits-archive@www.apache.org Delivered-To: apmail-drill-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 38C9B17CC5 for ; Thu, 26 Mar 2015 06:26:41 +0000 (UTC) Received: (qmail 66620 invoked by uid 500); 26 Mar 2015 06:26:41 -0000 Delivered-To: apmail-drill-commits-archive@drill.apache.org Received: (qmail 66549 invoked by uid 500); 26 Mar 2015 06:26:41 -0000 Mailing-List: contact commits-help@drill.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: commits@drill.apache.org Delivered-To: mailing list commits@drill.apache.org Received: (qmail 66399 invoked by uid 99); 26 Mar 2015 06:26:41 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 26 Mar 2015 06:26:40 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id A990DE2F05; Thu, 26 Mar 2015 06:26:40 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: jacques@apache.org To: commits@drill.apache.org Date: Thu, 26 Mar 2015 06:26:48 -0000 Message-Id: In-Reply-To: References: X-Mailer: ASF-Git Admin Mailer Subject: [09/10] drill git commit: DRILL-2574: SendingAccountor can suffer from lost updates DRILL-2574: SendingAccountor can suffer from lost updates SendingAccountor - atomically get and set the message count to wait for Project: http://git-wip-us.apache.org/repos/asf/drill/repo Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/db2e0321 Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/db2e0321 Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/db2e0321 Branch: refs/heads/0.8.0 Commit: db2e032150ff85bc3c0bd0761efc118bc57a18bf Parents: 26463d3 Author: Chris Westin Authored: Wed Mar 25 18:44:41 2015 -0700 Committer: Jacques Nadeau Committed: Wed Mar 25 21:11:23 2015 -0700 ---------------------------------------------------------------------- .../apache/drill/exec/physical/impl/SendingAccountor.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/drill/blob/db2e0321/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/SendingAccountor.java ---------------------------------------------------------------------- diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/SendingAccountor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/SendingAccountor.java index 8794188..21fc800 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/SendingAccountor.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/SendingAccountor.java @@ -27,10 +27,10 @@ import java.util.concurrent.atomic.AtomicInteger; * TODO: Need to update to use long for number of pending messages. */ public class SendingAccountor { - static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SendingAccountor.class); + private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SendingAccountor.class); private final AtomicInteger batchesSent = new AtomicInteger(0); - private Semaphore wait = new Semaphore(0); + private final Semaphore wait = new Semaphore(0); public void increment() { batchesSent.incrementAndGet(); @@ -42,8 +42,10 @@ public class SendingAccountor { public synchronized void waitForSendComplete() { try { - wait.acquire(batchesSent.get()); - batchesSent.set(0); + int waitForBatches; + while((waitForBatches = batchesSent.getAndSet(0)) != 0) { + wait.acquire(waitForBatches); + } } catch (InterruptedException e) { logger.warn("Failure while waiting for send complete.", e); // TODO InterruptedException