/ - Diff - D-Net - D-Net project tracking tool

« Previous | Next »

Revision 62288

Added by Michele Artini over 2 years ago

fixed a problem with blocked threads

     package eu.dnetlib.data.mdstore.modular.mongodb;
     import java.util.*;
     import java.util.concurrent.*;
     import java.util.ArrayList;
     import java.util.List;
     import java.util.Map;
     import java.util.Set;
     import java.util.concurrent.ArrayBlockingQueue;
     import java.util.concurrent.BlockingQueue;
     import java.util.concurrent.Callable;
     import java.util.concurrent.ExecutionException;
     import java.util.concurrent.ExecutorService;
     import java.util.concurrent.Executors;
     import java.util.concurrent.Future;
     import java.util.concurrent.TimeUnit;
     import java.util.regex.Pattern;
     import org.apache.commons.lang3.StringUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.bson.conversions.Bson;
     import org.springframework.beans.factory.annotation.Required;
     import com.google.common.base.Function;
     import com.google.common.collect.Iterables;
     import com.google.common.collect.Iterators;
-...
     import com.mongodb.client.MongoCollection;
     import com.mongodb.client.MongoDatabase;
     import com.mongodb.client.model.IndexOptions;
     import eu.dnetlib.data.mdstore.DocumentNotFoundException;
     import eu.dnetlib.data.mdstore.MDStoreServiceException;
     import eu.dnetlib.data.mdstore.modular.MDFormatDescription;
     import eu.dnetlib.data.mdstore.modular.RecordParser;
     import eu.dnetlib.data.mdstore.modular.connector.MDStore;
     import eu.dnetlib.enabling.resultset.ResultSetListener;
     import eu.dnetlib.enabling.resultset.client.ResultSetClient;
     import org.apache.commons.lang3.StringUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.bson.conversions.Bson;
     import org.springframework.beans.factory.annotation.Required;
     public class MongoMDStore implements MDStore {
-...
     	private final boolean discardRecords;
     	private String id;
     	private MongoDatabase mongoDatabase;
     	private final MongoDatabase mongoDatabase;
     	private MongoCollection<DBObject> collection;
     	private MongoCollection<DBObject> discardedCollection;
     	private List<MDFormatDescription> mdformats;
-...
     	private final IndexOptions options = new IndexOptions().background(true);
     	public MongoMDStore(final String id,
     			final MongoCollection<DBObject> collection,
     			final RecordParser recordParser,
     			final boolean discardRecords,
     			final MongoDatabase mongoDatabase) {
     		final MongoCollection<DBObject> collection,
     		final RecordParser recordParser,
     		final boolean discardRecords,
     		final MongoDatabase mongoDatabase) {
     		this.id = id;
     		this.mongoDatabase = mongoDatabase;
     		this.collection = collection;
-...
     		int countStored = 0;
     		final Callable<Integer> writer = () -> {
     			final MongoBulkWritesManager bulkWritesManager =
     					new MongoBulkWritesManager(collection, discardedCollection, mdformats, BULK_SIZE, recordParser, discardRecords);
     				new MongoBulkWritesManager(collection, discardedCollection, mdformats, BULK_SIZE, recordParser, discardRecords);
     			int count = 0;
     			while (true) {
     				try {
     					final Object record = queue.take();
     					final Object record = queue.poll(1, TimeUnit.HOURS);
     					if (record == null) {
     						log.fatal("retrieved a null object, probably the feeding is failed");
     						throw new IllegalStateException("retrieved a null object, probably the feeding wf is failed");
+    					}
     					if (record.equals(sentinel)) {
     						bulkWritesManager.flushBulks();
     						break;
-...
     			return count;
     		};
     		final ExecutorService executorService = Executors.newSingleThreadExecutor();
     		Future<Integer> storedCountInt = executorService.submit(writer);
     		final Future<Integer> storedCountInt = executorService.submit(writer);
     		try {
     			log.info("feeding mdstore " + id);
     			if (records != null) {
     				for (final String record : records) {
     					queue.put(record);
     					if (record != null) {
     						queue.put(record);
+    					}
+    				}
+    			}
     			queue.put(sentinel);
-...
     	 */
     	public void replace(final String grep, final String replace) {
     		final Pattern regex = Pattern.compile(grep, Pattern.MULTILINE);
     		BasicDBObject query = (BasicDBObject) QueryBuilder.start("body").regex(regex).get();
     		final BasicDBObject query = (BasicDBObject) QueryBuilder.start("body").regex(regex).get();
     		final FindIterable<DBObject> matches = collection.find(query, DBObject.class);
     		//final DBCursor matches = collection.find(QueryBuilder.start("body").regex(regex).get());
     		if (log.isDebugEnabled())
     		// final DBCursor matches = collection.find(QueryBuilder.start("body").regex(regex).get());
     		if (log.isDebugEnabled()) {
     			log.debug("FOUND: " + Lists.newArrayList(matches).size());
+    		}
     		for (final DBObject match : matches) {
     			final DBObject o = new BasicDBObject(match.toMap());
-...
+    	}
     	public ResultSetListener deliver(final String from, final String until, final String recordFilter, final Function<DBObject, String> serializer)
     			throws MDStoreServiceException {
     		final Pattern filter = (recordFilter != null) && (recordFilter.length() > 0) ? Pattern.compile(recordFilter, Pattern.MULTILINE) : null;
     		throws MDStoreServiceException {
     		final Pattern filter = recordFilter != null && recordFilter.length() > 0 ? Pattern.compile(recordFilter, Pattern.MULTILINE) : null;
     		return new MongoResultSetListener(collection, parseLong(from), parseLong(until), filter, serializer);	}
     		return new MongoResultSetListener(collection, parseLong(from), parseLong(until), filter, serializer);
+    	}
     	private Long parseLong(final String s) throws MDStoreServiceException {
     		if (StringUtils.isBlank(s)) {
     			return null;
+    		}
     		if (StringUtils.isBlank(s)) { return null; }
     		try {
     			return Long.valueOf(s);
     		} catch (NumberFormatException e) {
     		} catch (final NumberFormatException e) {
     			throw new MDStoreServiceException("Invalid date, expected java.lang.Long, or null", e);
+    		}
+    	}
-...
     	@Override
     	public String getRecord(final String recordId) throws DocumentNotFoundException {
     		final DBObject obj = collection.find(new BasicDBObject("id", recordId)).first();
     		if (obj == null || !obj.containsField("body")) throw new DocumentNotFoundException(String.format(
     				"The document with id '%s' does not exist in mdstore: '%s'", recordId, id));
     		if (obj == null || !obj.containsField("body")) {
     			throw new DocumentNotFoundException(String.format("The document with id '%s' does not exist in mdstore: '%s'", recordId, id));
+    		}
     		final String body = (String) obj.get("body");
     		if (body.trim().length() == 0) throw new DocumentNotFoundException(String.format("The document with id '%s' does not exist in mdstore: '%s'",
     				recordId, id));
     		if (body.trim().length() == 0) {
     			throw new DocumentNotFoundException(String.format("The document with id '%s' does not exist in mdstore: '%s'", recordId, id));
+    		}
     		return new SerializeMongoRecord().apply(obj);
+    	}
-...
     	public List<String> deliver(final String mdId, final int pageSize, final int offset, final Map<String, String> queryParam) {
     		final QueryBuilder query = QueryBuilder.start();
     		for (String key : queryParam.keySet()) {
     		for (final String key : queryParam.keySet()) {
     			query.and(key).regex(Pattern.compile(queryParam.get(key), Pattern.LITERAL));
+    		}
     		FindIterable<DBObject> dbObjects = offset > 0
     				? collection.find((Bson) query.get()).limit(pageSize).skip(offset)
     				: collection.find((Bson) query.get()).limit(pageSize);
     		final FindIterable<DBObject> dbObjects = offset > 0
     			? collection.find((Bson) query.get()).limit(pageSize).skip(offset)
     			: collection.find((Bson) query.get()).limit(pageSize);
     		queryParam.put("count", "" + collection.count((Bson) query.get()));
     		final List<String> result = new ArrayList<>();

     package eu.dnetlib.data.mdstore.modular.mongodb;
     import static org.junit.Assert.assertEquals;
     import java.util.concurrent.ArrayBlockingQueue;
     import java.util.concurrent.BlockingQueue;
     import java.util.concurrent.Callable;
     import java.util.concurrent.ExecutionException;
     import java.util.concurrent.ExecutorService;
     import java.util.concurrent.Executors;
     import java.util.concurrent.Future;
     import java.util.concurrent.TimeUnit;
     import org.junit.Ignore;
     import org.junit.Test;
     public class SimpleBlockingQueueThreadTest {
     	final ExecutorService executorService = Executors.newSingleThreadExecutor();
     	@Test
     	@Ignore
     	public void test_1() {
     		final int res = feed(1, 2, 3, 5);
     		assertEquals(4, res);
     		System.out.println("END");
+    	}
     	@Test
     	@Ignore
     	public void test_2() throws InterruptedException {
     		try {
     			// The term 0 produces a division by 0 exception
     			final int res = feed(1, 2, 0, 5);
     			assertEquals(4, res);
     		} catch (final Throwable e) {
     			Thread.sleep(15000);
+    		}
     		System.out.println("END");
+    	}
     	public int feed(final int... records) {
     		final BlockingQueue<Object> queue = new ArrayBlockingQueue<>(100);
     		final Object sentinel = new Object();
     		int countStored = 0;
     		final Callable<Integer> writer = () -> {
     			int count = 0;
     			while (true) {
     				try {
     					final Object record = queue.poll(10, TimeUnit.SECONDS);
     					if (record == null) {
     						System.err.println("- thread: retrieved a null object, probably the feeding is failed");
     						throw new IllegalStateException("retrieved a null object, probably the feeding wf is failed");
+    					}
     					if (record.equals(sentinel)) {
     						System.out.println("- thread: sentinel found");
     						break;
+    					}
     					count++;
     					System.out.println("- thread: processing object: " + record);
     				} catch (final InterruptedException e) {
     					System.err.println("- thread: Error - " + e.getMessage());
     					throw new IllegalStateException(e);
+    				}
+    			}
     			System.out.println("- thread: Total processed: " + count);
     			return count;
     		};
     		final Future<Integer> storedCountInt = executorService.submit(writer);
     		try {
     			for (final int record : records) {
     				queue.put(100 / record);
     				Thread.sleep(4000);
+    			}
     			queue.put(sentinel);
     			countStored = storedCountInt.get().intValue();
     		} catch (final InterruptedException | ExecutionException e) {
     			System.err.println("Error: " + e.getMessage());
     			throw new IllegalStateException(e);
+    		}
     		return countStored;
+    	}
+    }

Also available in: Unified diff

Project

General

Profile

D-Net

Revision 62288

Added by Michele Artini over 2 years ago