You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by mf...@apache.org on 2014/04/29 16:14:09 UTC
[1/3] git commit: Fixed logic bug
Repository: incubator-streams
Updated Branches:
refs/heads/master fdba6fff9 -> be58524bf
Fixed logic bug
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/be58524b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/be58524b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/be58524b
Branch: refs/heads/master
Commit: be58524bf019f69c32270325591beca09f8de8d6
Parents: 1d0d97b
Author: mfranklin <mf...@apache.org>
Authored: Tue Apr 29 10:12:48 2014 -0400
Committer: mfranklin <mf...@apache.org>
Committed: Tue Apr 29 10:13:59 2014 -0400
----------------------------------------------------------------------
.../main/java/org/apache/streams/mongo/MongoPersistWriter.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/be58524b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
index e60f375..a4a089a 100644
--- a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
+++ b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
@@ -31,7 +31,6 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(MongoPersistWriter.class);
private final static long MAX_WRITE_LATENCY = 1000;
- private final static long MIN_WRITE_LATENCY = 100;
protected volatile Queue<StreamsDatum> persistQueue;
@@ -153,7 +152,9 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
protected void flushIfNecessary() {
long lastLatency = System.currentTimeMillis() - lastWrite.get();
- if (insertBatch.size() % 100 == 0 || (insertBatch.size() > 0 && lastLatency > MIN_WRITE_LATENCY && lastLatency > MAX_WRITE_LATENCY)) {
+ //Flush iff the size > 0 AND the size is divisible by 100 or the time between now and the last flush is greater
+ //than the maximum desired latency
+ if (insertBatch.size() > 0 && (insertBatch.size() % 100 == 0 || lastLatency > MAX_WRITE_LATENCY)) {
try {
flush();
} catch (IOException e) {
[3/3] git commit: Added configurable timeout to Providers
Posted by mf...@apache.org.
Added configurable timeout to Providers
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/85510a76
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/85510a76
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/85510a76
Branch: refs/heads/master
Commit: 85510a760d6fa3ab4b8d6afd833d220575d30f1e
Parents: fdba6ff
Author: mfranklin <mf...@apache.org>
Authored: Tue Apr 29 08:49:21 2014 -0400
Committer: mfranklin <mf...@apache.org>
Committed: Tue Apr 29 10:13:59 2014 -0400
----------------------------------------------------------------------
.../streams/local/builders/LocalStreamBuilder.java | 11 +++++++++--
.../streams/local/builders/StreamComponent.java | 8 +++++++-
.../streams/local/tasks/StreamsProviderTask.java | 14 ++++++++++++--
3 files changed, 28 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/85510a76/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
index bf1abe6..8e688ba 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
@@ -22,6 +22,7 @@ import java.util.concurrent.TimeUnit;
*/
public class LocalStreamBuilder implements StreamBuilder {
+ public static final String TIMEOUT_KEY = "TIMEOUT";
private Map<String, StreamComponent> providers;
private Map<String, StreamComponent> components;
private Queue<StreamsDatum> queue;
@@ -150,7 +151,7 @@ public class LocalStreamBuilder implements StreamBuilder {
int tasks = comp.getNumTasks();
List<StreamsTask> compTasks = new LinkedList<StreamsTask>();
for(int i=0; i < tasks; ++i) {
- StreamsTask task = comp.createConnectedTask();
+ StreamsTask task = comp.createConnectedTask(getTimeout());
task.setStreamConfig(this.streamConfig);
this.executor.submit(task);
compTasks.add(task);
@@ -162,7 +163,7 @@ public class LocalStreamBuilder implements StreamBuilder {
streamsTasks.put(comp.getId(), compTasks);
}
for(StreamComponent prov : this.providers.values()) {
- StreamsTask task = prov.createConnectedTask();
+ StreamsTask task = prov.createConnectedTask(getTimeout());
task.setStreamConfig(this.streamConfig);
this.executor.submit(task);
provTasks.put(prov.getId(), (StreamsProviderTask) task);
@@ -176,6 +177,9 @@ public class LocalStreamBuilder implements StreamBuilder {
for(StreamsProviderTask task : provTasks.values()) {
isRunning = isRunning || task.isRunning();
}
+ for(StreamComponent task: components.values()) {
+ isRunning = isRunning || task.getInBoundQueue().size() > 0;
+ }
if(isRunning) {
Thread.sleep(3000);
}
@@ -295,5 +299,8 @@ public class LocalStreamBuilder implements StreamBuilder {
return (Queue<StreamsDatum>) SerializationUtil.cloneBySerialization(this.queue);
}
+ protected int getTimeout() {
+ return streamConfig != null && streamConfig.containsKey(TIMEOUT_KEY) ? (Integer)streamConfig.get(TIMEOUT_KEY) : 3000;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/85510a76/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
index f5e9978..5508cb8 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/StreamComponent.java
@@ -160,9 +160,11 @@ public class StreamComponent {
/**
* Creates a {@link org.apache.streams.local.tasks.StreamsTask} that is running a clone of this component whose
* inbound and outbound queues are appropriately connected to the parent and child nodes.
+ *
* @return StreamsTask for this component
+ * @param timeout The timeout to use in milliseconds for any tasks that support configurable timeout
*/
- public StreamsTask createConnectedTask() {
+ public StreamsTask createConnectedTask(int timeout) {
StreamsTask task;
if(this.processor != null) {
if(this.numTasks > 1) {
@@ -201,6 +203,10 @@ public class StreamComponent {
task = new StreamsProviderTask(prov, this.sequence);
else
task = new StreamsProviderTask(prov, this.dateRange[0], this.dateRange[1]);
+ //Adjust the timeout if necessary
+ if(timeout > 0) {
+ ((StreamsProviderTask)task).setTimeout(timeout);
+ }
for(Queue<StreamsDatum> q : this.outBound.values()) {
task.addOutputQueue(q);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/85510a76/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
----------------------------------------------------------------------
diff --git a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
index 5cf515c..add44bb 100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/StreamsProviderTask.java
@@ -42,6 +42,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
private Map<String, Object> config;
private AtomicBoolean isRunning;
+ private int timeout;
private int zeros = 0;
private DatumStatusCounter statusCounter = new DatumStatusCounter();
@@ -57,6 +58,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
this.type = Type.READ_CURRENT;
this.keepRunning = new AtomicBoolean(true);
this.isRunning = new AtomicBoolean(true);
+ this.timeout = DEFAULT_TIMEOUT_MS;
}
/**
@@ -70,6 +72,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
this.sequence = sequence;
this.keepRunning = new AtomicBoolean(true);
this.isRunning = new AtomicBoolean(true);
+ this.timeout = DEFAULT_TIMEOUT_MS;
}
/**
@@ -86,6 +89,11 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
this.dateRange[END] = end;
this.keepRunning = new AtomicBoolean(true);
this.isRunning = new AtomicBoolean(true);
+ this.timeout = DEFAULT_TIMEOUT_MS;
+ }
+
+ public void setTimeout(int timeout) {
+ this.timeout = timeout;
}
@Override
@@ -103,6 +111,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
this.config = config;
}
+
@Override
public void run() {
try {
@@ -112,7 +121,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
switch(this.type) {
case PERPETUAL: {
provider.startStream();
- while(this.keepRunning.get() == true) {
+ while(this.keepRunning.get()) {
try {
resultSet = provider.readCurrent();
if( resultSet.size() == 0 )
@@ -121,7 +130,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
zeros = 0;
}
flushResults(resultSet);
- if( zeros > (DEFAULT_TIMEOUT_MS / DEFAULT_SLEEP_TIME_MS))
+ if( zeros > (timeout / DEFAULT_SLEEP_TIME_MS))
this.keepRunning.set(false);
Thread.sleep(DEFAULT_SLEEP_TIME_MS);
} catch (InterruptedException e) {
@@ -149,6 +158,7 @@ public class StreamsProviderTask extends BaseStreamsTask implements DatumStatusC
}
}
+ @Override
public boolean isRunning() {
return this.isRunning.get();
}
[2/3] git commit: Fixed issues with locking and flushing
Posted by mf...@apache.org.
Fixed issues with locking and flushing
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/1d0d97b5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/1d0d97b5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/1d0d97b5
Branch: refs/heads/master
Commit: 1d0d97b54cb4b71a22d58f1377a0557fc350d889
Parents: 85510a7
Author: mfranklin <mf...@apache.org>
Authored: Tue Apr 29 09:57:53 2014 -0400
Committer: mfranklin <mf...@apache.org>
Committed: Tue Apr 29 10:13:59 2014 -0400
----------------------------------------------------------------------
.../streams/mongo/MongoPersistWriter.java | 169 ++++++++++++-------
1 file changed, 105 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/1d0d97b5/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
index 8d8d4b3..e60f375 100644
--- a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
+++ b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
@@ -3,6 +3,7 @@ package org.apache.streams.mongo;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
import com.mongodb.DB;
import com.mongodb.DBAddress;
import com.mongodb.DBCollection;
@@ -22,15 +23,21 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
-public class MongoPersistWriter implements StreamsPersistWriter, Runnable
-{
+public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(MongoPersistWriter.class);
+ private final static long MAX_WRITE_LATENCY = 1000;
+ private final static long MIN_WRITE_LATENCY = 100;
protected volatile Queue<StreamsDatum> persistQueue;
private ObjectMapper mapper = new ObjectMapper();
+ private volatile AtomicLong lastWrite = new AtomicLong(System.currentTimeMillis());
+ private ScheduledExecutorService backgroundFlushTask = Executors.newSingleThreadScheduledExecutor();
private MongoConfiguration config;
@@ -38,12 +45,14 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable
protected DBAddress dbaddress;
protected DBCollection collection;
- protected List<DBObject> insertBatch = new ArrayList<DBObject>();
+ protected List<DBObject> insertBatch = Lists.newArrayList();
+
+ protected final ReadWriteLock lock = new ReentrantReadWriteLock();
public MongoPersistWriter() {
Config config = StreamsConfigurator.config.getConfig("mongo");
this.config = MongoConfigurator.detectConfiguration(config);
- this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
+ this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
}
public MongoPersistWriter(Queue<StreamsDatum> persistQueue) {
@@ -52,69 +61,50 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable
this.persistQueue = persistQueue;
}
- private synchronized void connectToMongo()
- {
- try {
- dbaddress = new DBAddress(config.getHost(), config.getPort().intValue(), config.getDb());
- } catch (UnknownHostException e) {
- e.printStackTrace();
- return;
- }
-
- client = MongoClient.connect(dbaddress);
-
- if( !Strings.isNullOrEmpty(config.getUser()) && !Strings.isNullOrEmpty(config.getPassword()))
- client.authenticate(config.getUser(), config.getPassword().toCharArray());
-
- if( !client.collectionExists(config.getCollection())) {
- client.createCollection(config.getCollection(), null);
- };
+ public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
+ this.persistQueue = persistQueue;
+ }
- collection = client.getCollection(config.getCollection());
+ public Queue<StreamsDatum> getPersistQueue() {
+ return persistQueue;
}
-
+
@Override
public void write(StreamsDatum streamsDatum) {
- DBObject dbObject;
- if( streamsDatum.getDocument() instanceof String ) {
- dbObject = (DBObject) JSON.parse((String)streamsDatum.getDocument());
- } else {
- try {
- ObjectNode node = mapper.valueToTree(streamsDatum.getDocument());
- dbObject = (DBObject) JSON.parse(node.toString());
- } catch (Exception e) {
- e.printStackTrace();
- LOGGER.warn("Unsupported type: " + streamsDatum.getDocument().getClass());
- return;
- }
+ DBObject dbObject = prepareObject(streamsDatum);
+ if (dbObject != null) {
+ addToBatch(dbObject);
+ flushIfNecessary();
}
-
- insertBatch.add(dbObject);
-
- if( insertBatch.size() % 100 == 0)
- try {
- flush();
- } catch (IOException e) {
- e.printStackTrace();
- return;
- }
}
- public void flush() throws IOException
- {
- collection.insert(insertBatch);
+ public void flush() throws IOException {
+ try {
+ LOGGER.debug("Attempting to flush {} items to mongo", insertBatch.size());
+ lock.writeLock().lock();
+ collection.insert(insertBatch);
+ lastWrite.set(System.currentTimeMillis());
+ insertBatch = Lists.newArrayList();
+ } finally {
+ lock.writeLock().unlock();
+ }
+
}
- public synchronized void close() throws IOException
- {
+ public synchronized void close() throws IOException {
client.cleanCursors(true);
+ backgroundFlushTask.shutdownNow();
}
public void start() {
-
connectToMongo();
-
+ backgroundFlushTask.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ flushIfNecessary();
+ }
+ }, 0, MAX_WRITE_LATENCY * 2, TimeUnit.MILLISECONDS);
}
public void stop() {
@@ -131,18 +121,11 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable
}
}
- public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
- this.persistQueue = persistQueue;
- }
-
- public Queue<StreamsDatum> getPersistQueue() {
- return persistQueue;
- }
-
+ @Override
public void run() {
- while(true) {
- if( persistQueue.peek() != null ) {
+ while (true) {
+ if (persistQueue.peek() != null) {
try {
StreamsDatum entry = persistQueue.remove();
write(entry);
@@ -152,7 +135,8 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable
}
try {
Thread.sleep(new Random().nextInt(1));
- } catch (InterruptedException e) {}
+ } catch (InterruptedException e) {
+ }
}
}
@@ -166,4 +150,61 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable
public void cleanUp() {
stop();
}
+
+ protected void flushIfNecessary() {
+ long lastLatency = System.currentTimeMillis() - lastWrite.get();
+ if (insertBatch.size() % 100 == 0 || (insertBatch.size() > 0 && lastLatency > MIN_WRITE_LATENCY && lastLatency > MAX_WRITE_LATENCY)) {
+ try {
+ flush();
+ } catch (IOException e) {
+ LOGGER.error("Error writing to Mongo", e);
+ }
+ }
+ }
+
+ protected void addToBatch(DBObject dbObject) {
+ try {
+ lock.readLock().lock();
+ insertBatch.add(dbObject);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ protected DBObject prepareObject(StreamsDatum streamsDatum) {
+ DBObject dbObject = null;
+ if (streamsDatum.getDocument() instanceof String) {
+ dbObject = (DBObject) JSON.parse((String) streamsDatum.getDocument());
+ } else {
+ try {
+ ObjectNode node = mapper.valueToTree(streamsDatum.getDocument());
+ dbObject = (DBObject) JSON.parse(node.toString());
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOGGER.error("Unsupported type: " + streamsDatum.getDocument().getClass(), e);
+ }
+ }
+ return dbObject;
+ }
+
+ private synchronized void connectToMongo() {
+ try {
+ dbaddress = new DBAddress(config.getHost(), config.getPort().intValue(), config.getDb());
+ } catch (UnknownHostException e) {
+ e.printStackTrace();
+ return;
+ }
+
+ client = MongoClient.connect(dbaddress);
+
+ if (!Strings.isNullOrEmpty(config.getUser()) && !Strings.isNullOrEmpty(config.getPassword()))
+ client.authenticate(config.getUser(), config.getPassword().toCharArray());
+
+ if (!client.collectionExists(config.getCollection())) {
+ client.createCollection(config.getCollection(), null);
+ }
+ ;
+
+ collection = client.getCollection(config.getCollection());
+ }
}