GitHub PullRequestReviewCommentEvent at 2014-05-14 08:35:35
Repository: hazelcast
Actor: ahmetmircik
Payload: {"comment":{"url":"https://api.github.com/repos/hazelcast/hazelcast/pulls/comments/12625886","id":12625886,"diff_hunk":"@@ -0,0 +1,331 @@\n+/*\n+* Copyright (c) 2008-2013, Hazelcast, Inc. All Rights Reserved.\n+*\n+* Licensed under the Apache License, Version 2.0 (the \"License\");\n+* you may not use this file except in compliance with the License.\n+* You may obtain a copy of the License at\n+*\n+* http://www.apache.org/licenses/LICENSE-2.0\n+*\n+* Unless required by applicable law or agreed to in writing, software\n+* distributed under the License is distributed on an \"AS IS\" BASIS,\n+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+* See the License for the specific language governing permissions and\n+* limitations under the License.\n+*/\n+\n+package com.hazelcast.map.writebehind;\n+\n+\n+import com.hazelcast.cluster.ClusterService;\n+import com.hazelcast.logging.ILogger;\n+import com.hazelcast.map.MapService;\n+import com.hazelcast.map.MapStoreWrapper;\n+import com.hazelcast.map.PartitionContainer;\n+import com.hazelcast.map.RecordStore;\n+import com.hazelcast.map.writebehind.store.MapStoreManager;\n+import com.hazelcast.map.writebehind.store.MapStoreManagers;\n+import com.hazelcast.map.writebehind.store.StoreListener;\n+import com.hazelcast.nio.Address;\n+import com.hazelcast.nio.serialization.Data;\n+import com.hazelcast.partition.InternalPartition;\n+import com.hazelcast.partition.InternalPartitionService;\n+import com.hazelcast.spi.ExecutionService;\n+import com.hazelcast.spi.NodeEngine;\n+import com.hazelcast.util.executor.ExecutorType;\n+\n+import java.util.ArrayList;\n+import java.util.Collection;\n+import java.util.Collections;\n+import java.util.Comparator;\n+import java.util.HashMap;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.concurrent.ScheduledExecutorService;\n+import java.util.concurrent.TimeUnit;\n+\n+/**\n+ * TODO add parallel sort.\n+ * <p/>\n+ * Write behind queue manager.\n+ */\n+class WriteBehindQueueManager implements WriteBehindManager {\n+\n+    private static final String EXECUTOR_NAME_PREFIX = \"hz:scheduled:mapstore:\";\n+\n+    private static final int EXECUTOR_DEFAULT_QUEUE_CAPACITY = 10000;\n+\n+    private static final Comparator<DelayedEntry> DELAYED_ENTRY_COMPARATOR = new Comparator<DelayedEntry>() {\n+        @Override\n+        public int compare(DelayedEntry o1, DelayedEntry o2) {\n+            final long s1 = o1.getStoreTime();\n+            final long s2 = o2.getStoreTime();\n+            return (s1 < s2) ? -1 : ((s1 == s2) ? 0 : 1);\n+        }\n+    };\n+\n+    private final ScheduledExecutorService scheduledExecutor;\n+\n+    private final StoreProcessor processor;\n+\n+    private final MapService mapService;\n+\n+    private final String executorName;\n+\n+    private final MapStoreManager<DelayedEntry> mapStoreManager;\n+\n+    private final List<StoreListener> listeners;\n+\n+    private final ILogger logger;\n+\n+    WriteBehindQueueManager(String mapName, MapService mapService, MapStoreWrapper storeWrapper) {\n+        this.scheduledExecutor = getScheduledExecutorService(mapName, mapService);\n+        this.mapService = mapService;\n+        this.logger = mapService.getNodeEngine().getLogger(WriteBehindQueueManager.class);\n+        this.executorName = EXECUTOR_NAME_PREFIX + mapName;\n+        this.listeners = new ArrayList<StoreListener>(2);\n+        this.mapStoreManager = MapStoreManagers.newMapStoreManager(mapService, storeWrapper, listeners);\n+        this.processor = new StoreProcessor(mapName, mapService, mapStoreManager, this);\n+    }\n+\n+    @Override\n+    public void start() {\n+        scheduledExecutor.scheduleAtFixedRate(processor, 1, 1, TimeUnit.SECONDS);\n+    }\n+\n+    @Override\n+    public void stop() {\n+        mapService.getNodeEngine().getExecutionService().shutdownExecutor(executorName);\n+    }\n+\n+    @Override\n+    public void reset() {\n+    }\n+\n+    @Override\n+    public void addStoreListener(StoreListener storeListener) {\n+        listeners.add(storeListener);\n+    }\n+\n+    @Override\n+    public Collection<Data> flush(WriteBehindQueue<DelayedEntry> queue) {\n+        if (queue.size() == 0) {\n+            return Collections.emptyList();\n+        }\n+        final List<DelayedEntry> sortedDelayedEntries = queue.fetchAndRemoveAll();\n+        Collections.sort(sortedDelayedEntries, DELAYED_ENTRY_COMPARATOR);\n+        final Map<Integer, Collection<DelayedEntry>> failedStoreOpPerPartition\n+                = new HashMap<Integer, Collection<DelayedEntry>>();\n+        mapStoreManager.process(sortedDelayedEntries, failedStoreOpPerPartition);\n+        if (failedStoreOpPerPartition.size() > 0) {\n+            printErrorLog(failedStoreOpPerPartition);\n+        }\n+        queue.clear();\n+        return getDataKeys(sortedDelayedEntries);\n+    }\n+\n+    @Override\n+    public ScheduledExecutorService getScheduler() {\n+        return scheduledExecutor;\n+    }\n+\n+    private void printErrorLog(Map<Integer, Collection<DelayedEntry>> failedsPerPartition) {\n+        int size = 0;\n+        final Collection<Collection<DelayedEntry>> values = failedsPerPartition.values();\n+        for (Collection<DelayedEntry> value : values) {\n+            size += value.size();\n+        }\n+        final String logMessage = String.format(\"Map store flush operation can not be done for %d entries\", size);\n+        logger.severe(logMessage);\n+    }\n+\n+    private static List<DelayedEntry> filterLessThanOrEqualToTime(WriteBehindQueue<DelayedEntry> queue,\n+                                                                  long time, TimeUnit unit) {\n+        if (queue == null || queue.size() == 0) {\n+            return Collections.emptyList();\n+        }\n+        final long timeInNanos = unit.toNanos(time);\n+        List<DelayedEntry> delayedEntries = Collections.emptyList();\n+        DelayedEntry e;\n+        int i = 0;\n+        while ((e = queue.get(i)) != null) {\n+            if (i == 0) {\n+                // init when needed.\n+                delayedEntries = new ArrayList<DelayedEntry>();\n+            }\n+            if (e.getStoreTime() <= timeInNanos) {\n+                delayedEntries.add(e);\n+            }\n+            i++;\n+        }\n+        return delayedEntries;\n+    }\n+\n+    private static void removeProcessed(WriteBehindQueue<DelayedEntry> queue, int numberOfEntriesProcessed) {\n+        if (queue == null || queue.size() == 0 || numberOfEntriesProcessed < 1) {\n+            return;\n+        }\n+        for (int j = 0; j < numberOfEntriesProcessed; j++) {\n+            queue.removeFirst();\n+        }\n+    }\n+\n+    private List<Data> getDataKeys(final List<DelayedEntry> sortedDelayedEntries) {\n+        if (sortedDelayedEntries == null || sortedDelayedEntries.isEmpty()) {\n+            return Collections.emptyList();\n+        }\n+        final List<Data> keys = new ArrayList<Data>(sortedDelayedEntries.size());\n+        for (DelayedEntry entry : sortedDelayedEntries) {\n+            // TODO Key type always should be Data. No need to call mapService.toData but check first if it is.\n+            keys.add(mapService.toData(entry.getKey()));\n+        }\n+        return keys;\n+    }\n+\n+    private ScheduledExecutorService getScheduledExecutorService(String mapName, MapService mapService) {\n+        final NodeEngine nodeEngine = mapService.getNodeEngine();\n+        final ExecutionService executionService = nodeEngine.getExecutionService();\n+        final String executorName = EXECUTOR_NAME_PREFIX + mapName;\n+        executionService.register(executorName, 1, EXECUTOR_DEFAULT_QUEUE_CAPACITY, ExecutorType.CACHED);\n+        return executionService.getScheduledExecutor(executorName);\n+    }\n+\n+    private static void removeProcessedEntries(MapService mapService, String mapName,\n+                                               Map<Integer, Integer> partitionToEntryCountHolder, Map<Integer,\n+            Collection<DelayedEntry>> failedsPerPartition) {\n+        for (Map.Entry<Integer, Integer> entry : partitionToEntryCountHolder.entrySet()) {\n+            final Integer partitionId = entry.getKey();\n+            final PartitionContainer partitionContainer = mapService.getPartitionContainer(partitionId);\n+            final RecordStore recordStore = partitionContainer.getExistingRecordStore(mapName);\n+            if (recordStore == null) {\n+                continue;\n+            }\n+            final WriteBehindQueue<DelayedEntry> queue = recordStore.getWriteBehindQueue();\n+            removeProcessed(queue, partitionToEntryCountHolder.get(partitionId));\n+            final Collection<DelayedEntry> faileds = failedsPerPartition.get(partitionId);\n+            if (faileds == null || faileds.isEmpty()) {\n+                continue;\n+            }\n+            queue.addFront(faileds);\n+\n+        }\n+    }\n+\n+    /**\n+     * Process store operations.\n+     */\n+    private static final class StoreProcessor implements Runnable {\n+\n+        private long lastRunTimeInNanos = nanoNow();\n+\n+        private final String mapName;\n+\n+        private final MapService mapService;\n+\n+        private final MapStoreManager mapStoreManager;\n+\n+        private final WriteBehindManager writeBehindManager;\n+\n+        private final long backupWorkIntervalTimeInNanos;\n+\n+        private StoreProcessor(String mapName, MapService mapService,\n+                               MapStoreManager mapStoreManager, WriteBehindManager manager) {\n+            this.mapName = mapName;\n+            this.mapService = mapService;\n+            this.mapStoreManager = mapStoreManager;\n+            this.writeBehindManager = manager;\n+            this.backupWorkIntervalTimeInNanos = getReplicaWaitTimeInNanos();\n+\n+        }\n+\n+        private long getReplicaWaitTimeInNanos() {\n+            return TimeUnit.SECONDS.toNanos(mapService.getNodeEngine().getGroupProperties()\n+                    .MAP_REPLICA_WAIT_SECONDS_FOR_SCHEDULED_TASKS.getInteger());\n+        }\n+\n+        @Override\n+        public void run() {","path":"hazelcast/src/main/java/com/hazelcast/map/writebehind/WriteBehindQueueManager.java","position":247,"original_position":247,"commit_id":"5ce249238b516ecb8616375b0605c8c1a208ad40","original_commit_id":"5ce249238b516ecb8616375b0605c8c1a208ad40","user":{"login":"ahmetmircik","id":2960360,"avatar_url":"https://avatars.githubusercontent.com/u/2960360?","gravatar_id":"9d7b99f62eea222bf52b9cd4764c80e9","url":"https://api.github.com/users/ahmetmircik","html_url":"https://github.com/ahmetmircik","followers_url":"https://api.github.com/users/ahmetmircik/followers","following_url":"https://api.github.com/users/ahmetmircik/following{/other_user}","gists_url":"https://api.github.com/users/ahmetmircik/gists{/gist_id}","starred_url":"https://api.github.com/users/ahmetmircik/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ahmetmircik/subscriptions","organizations_url":"https://api.github.com/users/ahmetmircik/orgs","repos_url":"https://api.github.com/users/ahmetmircik/repos","events_url":"https://api.github.com/users/ahmetmircik/events{/privacy}","received_events_url":"https://api.github.com/users/ahmetmircik/received_events","type":"User","site_admin":false},"body":"Already on todo list.","created_at":"2014-05-14T08:35:35Z","updated_at":"2014-05-14T08:35:35Z","html_url":"https://github.com/hazelcast/hazelcast/pull/2449#discussion_r12625886","pull_request_url":"https://api.github.com/repos/hazelcast/hazelcast/pulls/2449","_links":{"self":{"href":"https://api.github.com/repos/hazelcast/hazelcast/pulls/comments/12625886"},"html":{"href":"https://github.com/hazelcast/hazelcast/pull/2449#discussion_r12625886"},"pull_request":{"href":"https://api.github.com/repos/hazelcast/hazelcast/pulls/2449"}}}}