You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@streams.apache.org by rbnks <gi...@git.apache.org> on 2014/10/08 20:08:32 UTC
[GitHub] incubator-streams pull request: Streams 190
GitHub user rbnks opened a pull request:
https://github.com/apache/incubator-streams/pull/99
Streams 190
Creates a throughput queue that allow JMX monitoring of the queues inside of local stream builder.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/rbnks/incubator-streams STREAMS-190
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-streams/pull/99.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #99
----
commit c483941405f3ef1792c35ea7e9495980eb6e8b96
Author: Ryan Ebanks <ry...@raveldata.com>
Date: 2014-09-30T19:50:52Z
Merge pull request #5 from apache/master
apache master merge 2014/09/30
commit 950330952a4aab51e7f273c27e8dad2eafeed0b5
Author: Ryan Ebanks <ry...@gmail.com>
Date: 2014-10-06T20:38:37Z
Added ThroughputQueue and Tests
commit 50906edf8af28159c13cc251c9e7bf12d4877799
Author: Ryan Ebanks <ry...@gmail.com>
Date: 2014-10-06T20:43:43Z
Fixed javadocs
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18672624
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,377 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock takeCountsLock;
+ private AtomicLong elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
--- End diff --
Yes, but they get accessed at the same time a lot of the time, so instead of acquiring three locks, I tried to simplify it under on lock. I can change this if the have Atomics is preferred.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18608539
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,384 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
--- End diff --
It could be an Atomic along as every time its accessed is an atomic operation. At the time of creation I did not think that was going to be case. If you would like I can change it to make it an AtomicLong. But basically I thought I would need more control over the locking than an Atomic allows, but was incorrect.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18621693
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,377 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock takeCountsLock;
+ private AtomicLong elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = new AtomicLong(0);
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if(this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+
+ @Override
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int remainingCapacity() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E remove() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E poll() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E element() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E peek() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int size() {
+ return this.underlyingQueue.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.underlyingQueue.isEmpty();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public Object[] toArray() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void clear() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public long getCurrentSize() {
+ long size = -1;
+ try {
--- End diff --
If you convert the other value to a AtomicLong, I don't think you need to lock here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18669084
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,377 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock takeCountsLock;
+ private AtomicLong elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = new AtomicLong(0);
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
--- End diff --
There not used in the current implementation. Can be easily added and tested.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18604876
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueueMXBean.java ---
@@ -0,0 +1,48 @@
+package org.apache.streams.local.queues;
+
+import javax.management.MXBean;
+
+/**
+ * MXBean capable queue that monitors the throughput of the queue
+ */
+public interface ThroughputQueueMXBean {
+
+ /**
+ * Returns the number of items on the queue.
+ * @return number of items on queue
+ */
+ public long getCurrentSize();
+
+ /**
+ * Get the average time an item spends in queue in milliseconds
+ * @return average time an item spends in queue in milliseconds
+ */
+ public double getAvgWait();
+
+ /**
+ * Get the maximum time an item has spent on the queue before being removed from the queue.
+ * @return the maximum time an item has spent on the queue
+ */
+ public long getMaxWait();
+
+ /**
+ * Get the number of items that have been removed from this queue
+ * @return number of items that have been removed from the queue
+ */
+ public long getRemoved();
+
+ /**
+ * Get the number of items that have been added to the queue
+ * @return number of items that have been added to the queue
+ */
+ public long getAdded();
+
+ /**
+ * Get the the throughput of the queue measured by the time the queue has been active divided by
--- End diff --
Documentation nit: Shouldn't this read: "Get the throughput of the queue measured by the number of items removed from the queue divided by the amount of time the queue has been active."?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18607089
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,384 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
--- End diff --
Why not use AtomicLong?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18731291
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,498 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ * <p/>
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private AtomicLong elementsAdded;
+ private AtomicLong elementsRemoved;
+ private AtomicLong startTime;
+ private AtomicLong totalQueueTime;
+ private long maxQueuedTime;
+ private volatile boolean active;
+ private ReadWriteLock maxQueueTimeLock;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ *
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ *
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ *
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if (maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = new AtomicLong(0);
+ this.elementsRemoved = new AtomicLong(0);
+ this.startTime = new AtomicLong(-1);
+ this.active = false;
+ this.maxQueuedTime = 0;
+ this.maxQueueTimeLock = new ReentrantReadWriteLock();
+ this.totalQueueTime = new AtomicLong(0);
+ if (id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException | InstanceAlreadyExistsException | MBeanRegistrationException | NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ if (this.underlyingQueue.add(new ThroughputElement<E>(e))) {
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime.set(System.currentTimeMillis());
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean offer(E e) {
+ if (this.underlyingQueue.offer(new ThroughputElement<E>(e))) {
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime.set(System.currentTimeMillis());
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime.set(System.currentTimeMillis());
+ this.active = true;
+ }
+ }
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if (this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime.set(System.currentTimeMillis());
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ this.elementsRemoved.incrementAndGet();
--- End diff --
This looks like the same logic as poll. Could this be abstracted to a single method?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-streams/pull/99
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18606180
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,382 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = 0;
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.putCountsLock = new ReentrantReadWriteLock();
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if(this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+
+ @Override
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int remainingCapacity() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E remove() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E poll() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E element() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E peek() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int size() {
+ return this.underlyingQueue.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.underlyingQueue.isEmpty();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public Object[] toArray() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void clear() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public long getCurrentSize() {
+ long size = -1;
+ try {
+ this.putCountsLock.readLock().lock();
+ try {
+ this.takeCountsLock.readLock().lock();
+ size = this.elementsAdded - this.elementsRemoved;
+ } finally {
--- End diff --
Yes, an exception would be thrown. Try/Finally blocks just says that if an exception is thrown in that block, before throwing the exception run the code that is in the finally block first. The finally block attempts to release the lock it may have obtained in the try block. This is to avoid having a situation where this thread obtains a lock, gets an exception, and dies while still holding a lock.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18731362
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,498 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ * <p/>
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private AtomicLong elementsAdded;
+ private AtomicLong elementsRemoved;
+ private AtomicLong startTime;
+ private AtomicLong totalQueueTime;
+ private long maxQueuedTime;
+ private volatile boolean active;
+ private ReadWriteLock maxQueueTimeLock;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ *
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ *
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ *
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if (maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = new AtomicLong(0);
+ this.elementsRemoved = new AtomicLong(0);
+ this.startTime = new AtomicLong(-1);
+ this.active = false;
+ this.maxQueuedTime = 0;
+ this.maxQueueTimeLock = new ReentrantReadWriteLock();
+ this.totalQueueTime = new AtomicLong(0);
+ if (id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException | InstanceAlreadyExistsException | MBeanRegistrationException | NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ if (this.underlyingQueue.add(new ThroughputElement<E>(e))) {
+ this.elementsAdded.incrementAndGet();
+ synchronized (this) {
--- End diff --
Could this be abstracted to a common method
`
this.elementsAdded.incrementAndGet();
synchronized (this) {
if (!this.active) {
this.startTime.set(System.currentTimeMillis());
this.active = true;
} }`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18604582
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,382 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = 0;
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.putCountsLock = new ReentrantReadWriteLock();
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if(this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+
+ @Override
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int remainingCapacity() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E remove() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E poll() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E element() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E peek() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int size() {
+ return this.underlyingQueue.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.underlyingQueue.isEmpty();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public Object[] toArray() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void clear() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public long getCurrentSize() {
+ long size = -1;
+ try {
+ this.putCountsLock.readLock().lock();
+ try {
+ this.takeCountsLock.readLock().lock();
+ size = this.elementsAdded - this.elementsRemoved;
+ } finally {
--- End diff --
If we were to get an exception here, would that bubble up and get reported?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18621552
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,377 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock takeCountsLock;
+ private AtomicLong elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
--- End diff --
glad to see the AtomicLong for added, couldn't elementsRemoved, and other fields be atomic too?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18621496
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,377 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock takeCountsLock;
+ private AtomicLong elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = new AtomicLong(0);
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
--- End diff --
Why not support add and offer?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18606388
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,382 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = 0;
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.putCountsLock = new ReentrantReadWriteLock();
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if(this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+
+ @Override
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int remainingCapacity() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E remove() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E poll() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E element() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E peek() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int size() {
+ return this.underlyingQueue.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.underlyingQueue.isEmpty();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public Object[] toArray() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void clear() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public long getCurrentSize() {
+ long size = -1;
+ try {
+ this.putCountsLock.readLock().lock();
+ try {
+ this.takeCountsLock.readLock().lock();
+ size = this.elementsAdded - this.elementsRemoved;
+ } finally {
+ this.takeCountsLock.readLock().unlock();
+ }
+ } finally {
+ this.putCountsLock.readLock().unlock();
+ }
+ return size;
+ }
+
+ @Override
+ public double getAvgWait() {
+ double avg = -1.0;
+ try {
+ this.takeCountsLock.readLock().lock();
+ avg = (double) this.totalQueueTime / (double) this.elementsRemoved;
+ } finally {
+ this.takeCountsLock.readLock().unlock();
+ }
+ return avg;
+ }
+
+ @Override
+ public long getMaxWait() {
+ ThroughputElement<E> e = this.underlyingQueue.peek();
+ long max = -1;
+ try {
+ this.takeCountsLock.readLock().lock();
+ if (e != null && e.getWaited() > this.maxQueuedTime) {
+ max = e.getWaited();
--- End diff --
No this is only suppose to get the current max wait time. So it tracks the longest wait time of removed elements. Then when max wait gets called it peeks at the head of the queue and sees how long that element has been waiting. If the element has been waiting longer than any previously de-queued elements it is currently the max wait. But we don't need to set it to max wait because that will happen when it gets de-queued with its actual wait time since it is still waiting.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18606496
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueueMXBean.java ---
@@ -0,0 +1,48 @@
+package org.apache.streams.local.queues;
+
+import javax.management.MXBean;
+
+/**
+ * MXBean capable queue that monitors the throughput of the queue
+ */
+public interface ThroughputQueueMXBean {
+
+ /**
+ * Returns the number of items on the queue.
+ * @return number of items on queue
+ */
+ public long getCurrentSize();
+
+ /**
+ * Get the average time an item spends in queue in milliseconds
+ * @return average time an item spends in queue in milliseconds
+ */
+ public double getAvgWait();
+
+ /**
+ * Get the maximum time an item has spent on the queue before being removed from the queue.
+ * @return the maximum time an item has spent on the queue
+ */
+ public long getMaxWait();
+
+ /**
+ * Get the number of items that have been removed from this queue
+ * @return number of items that have been removed from the queue
+ */
+ public long getRemoved();
+
+ /**
+ * Get the number of items that have been added to the queue
+ * @return number of items that have been added to the queue
+ */
+ public long getAdded();
+
+ /**
+ * Get the the throughput of the queue measured by the time the queue has been active divided by
--- End diff --
You are correct. Will fix.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on the pull request:
https://github.com/apache/incubator-streams/pull/99#issuecomment-58722042
@mfranklin Good call. That was some lazy programming. Fixed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18711767
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,377 @@
+/*
+ * 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
+ *
+ * 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.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock takeCountsLock;
+ private AtomicLong elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
--- End diff --
Atomics don't actually lock. They do a native compare + swap on changes
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18604651
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,382 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = 0;
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.putCountsLock = new ReentrantReadWriteLock();
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if(this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+
+ @Override
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int remainingCapacity() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E remove() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E poll() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E element() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E peek() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int size() {
+ return this.underlyingQueue.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.underlyingQueue.isEmpty();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public Object[] toArray() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void clear() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public long getCurrentSize() {
+ long size = -1;
+ try {
+ this.putCountsLock.readLock().lock();
+ try {
+ this.takeCountsLock.readLock().lock();
+ size = this.elementsAdded - this.elementsRemoved;
+ } finally {
+ this.takeCountsLock.readLock().unlock();
+ }
+ } finally {
+ this.putCountsLock.readLock().unlock();
+ }
+ return size;
+ }
+
+ @Override
+ public double getAvgWait() {
+ double avg = -1.0;
+ try {
+ this.takeCountsLock.readLock().lock();
+ avg = (double) this.totalQueueTime / (double) this.elementsRemoved;
+ } finally {
+ this.takeCountsLock.readLock().unlock();
+ }
+ return avg;
+ }
+
+ @Override
+ public long getMaxWait() {
+ ThroughputElement<E> e = this.underlyingQueue.peek();
+ long max = -1;
+ try {
+ this.takeCountsLock.readLock().lock();
+ if (e != null && e.getWaited() > this.maxQueuedTime) {
+ max = e.getWaited();
--- End diff --
Do we not need to assign e.getWaited() to this.maxQueuedTime here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/99#discussion_r18606518
--- Diff: streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/queues/ThroughputQueue.java ---
@@ -0,0 +1,382 @@
+package org.apache.streams.local.queues;
+
+import net.jcip.annotations.GuardedBy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import javax.management.*;
+import java.lang.management.ManagementFactory;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * A {@link java.util.concurrent.BlockingQueue} implementation that allows the measure measurement of how
+ * data flows through the queue. Is also a {@code MBean} so the flow statistics can be viewed through
+ * JMX. Registration of the bean happens whenever a constructor receives a non-null id.
+ *
+ * !!! Warning !!!
+ * Only the necessary methods for the local streams runtime are implemented. All other methods throw a
+ * {@link sun.reflect.generics.reflectiveObjects.NotImplementedException}.
+ */
+public class ThroughputQueue<E> implements BlockingQueue<E>, ThroughputQueueMXBean {
+
+ public static final String NAME_TEMPLATE = "org.apache.streams.local:type=ThroughputQueue,name=%s";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThroughputQueue.class);
+
+ private BlockingQueue<ThroughputElement<E>> underlyingQueue;
+ private ReadWriteLock putCountsLock;
+ private ReadWriteLock takeCountsLock;
+ @GuardedBy("putCountsLock")
+ private long elementsAdded;
+ @GuardedBy("takeCountsLock")
+ private long elementsRemoved;
+ @GuardedBy("this")
+ private long startTime;
+ @GuardedBy("takeCountsLock")
+ private long totalQueueTime;
+ @GuardedBy("takeCountsLock")
+ private long maxQueuedTime;
+ private volatile boolean active;
+
+ /**
+ * Creates an unbounded, unregistered {@code ThroughputQueue}
+ */
+ public ThroughputQueue() {
+ this(-1, null);
+ }
+
+ /**
+ * Creates a bounded, unregistered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ */
+ public ThroughputQueue(int maxSize) {
+ this(maxSize, null);
+ }
+
+ /**
+ * Creates an unbounded, registered {@code ThroughputQueue}
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(String id) {
+ this(-1, id);
+ }
+
+ /**
+ * Creates a bounded, registered {@code ThroughputQueue}
+ * @param maxSize maximum capacity of queue, if maxSize < 1 then unbounded
+ * @param id unique id for this queue to be registered with. if id == NULL then not registered
+ */
+ public ThroughputQueue(int maxSize, String id) {
+ if(maxSize < 1) {
+ this.underlyingQueue = new LinkedBlockingQueue<>();
+ } else {
+ this.underlyingQueue = new LinkedBlockingQueue<>(maxSize);
+ }
+ this.elementsAdded = 0;
+ this.elementsRemoved = 0;
+ this.startTime = -1;
+ this.putCountsLock = new ReentrantReadWriteLock();
+ this.takeCountsLock = new ReentrantReadWriteLock();
+ this.active = false;
+ this.maxQueuedTime = -1;
+ if(id != null) {
+ try {
+ ObjectName name = new ObjectName(String.format(NAME_TEMPLATE, id));
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ mbs.registerMBean(this, name);
+ } catch (MalformedObjectNameException|InstanceAlreadyExistsException|MBeanRegistrationException|NotCompliantMBeanException e) {
+ LOGGER.error("Failed to register MXBean : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public boolean add(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean offer(E e) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void put(E e) throws InterruptedException {
+ this.underlyingQueue.put(new ThroughputElement<E>(e));
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+
+ }
+
+ @Override
+ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+ if(this.underlyingQueue.offer(new ThroughputElement<E>(e), timeout, unit)) {
+ try {
+ this.putCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsAdded;
+ } finally {
+ this.putCountsLock.writeLock().unlock();
+ }
+ synchronized (this) {
+ if (!this.active) {
+ this.startTime = System.currentTimeMillis();
+ this.active = true;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ ThroughputElement<E> e = this.underlyingQueue.take();
+ try {
+ this.takeCountsLock.writeLock().lockInterruptibly();
+ ++this.elementsRemoved;
+ Long queueTime = e.getWaited();
+ this.totalQueueTime += queueTime;
+ if(this.maxQueuedTime < queueTime) {
+ this.maxQueuedTime = queueTime;
+ }
+ } finally {
+ this.takeCountsLock.writeLock().unlock();
+ }
+ return e.getElement();
+ }
+
+ @Override
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int remainingCapacity() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E remove() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E poll() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E element() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public E peek() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public int size() {
+ return this.underlyingQueue.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.underlyingQueue.isEmpty();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public Object[] toArray() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public void clear() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public long getCurrentSize() {
+ long size = -1;
+ try {
+ this.putCountsLock.readLock().lock();
+ try {
+ this.takeCountsLock.readLock().lock();
+ size = this.elementsAdded - this.elementsRemoved;
+ } finally {
+ this.takeCountsLock.readLock().unlock();
+ }
+ } finally {
+ this.putCountsLock.readLock().unlock();
+ }
+ return size;
+ }
+
+ @Override
+ public double getAvgWait() {
+ double avg = -1.0;
+ try {
+ this.takeCountsLock.readLock().lock();
+ avg = (double) this.totalQueueTime / (double) this.elementsRemoved;
+ } finally {
+ this.takeCountsLock.readLock().unlock();
+ }
+ return avg;
+ }
+
+ @Override
+ public long getMaxWait() {
+ ThroughputElement<E> e = this.underlyingQueue.peek();
+ long max = -1;
+ try {
+ this.takeCountsLock.readLock().lock();
+ if (e != null && e.getWaited() > this.maxQueuedTime) {
+ max = e.getWaited();
--- End diff --
Ah missed that bit in the take() method. Sounds good
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas commented on the pull request:
https://github.com/apache/incubator-streams/pull/99#issuecomment-58414009
:+1:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on the pull request:
https://github.com/apache/incubator-streams/pull/99#issuecomment-58677987
Added new tests and implementations. Also converted most locking to Atomics
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 190
Posted by rbnks <gi...@git.apache.org>.
Github user rbnks commented on the pull request:
https://github.com/apache/incubator-streams/pull/99#issuecomment-58427891
Changed to variable to AtomicLong per feedback.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---