<commit_before>/*
 * Copyright (c) 2008-2013, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * 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 com.hazelcast.map;

import com.hazelcast.config.MapConfig;
import com.hazelcast.config.MapStoreConfig;
import com.hazelcast.config.PartitioningStrategyConfig;
import com.hazelcast.config.WanReplicationRef;
import com.hazelcast.core.MapLoaderLifecycleSupport;
import com.hazelcast.core.MapStoreFactory;
import com.hazelcast.core.PartitioningStrategy;
import com.hazelcast.map.eviction.ReachabilityHandlerChain;
import com.hazelcast.map.eviction.ReachabilityHandlers;
import com.hazelcast.map.merge.MapMergePolicy;
import com.hazelcast.map.operation.ClearExpiredOperation;
import com.hazelcast.map.record.DataRecordFactory;
import com.hazelcast.map.record.ObjectRecordFactory;
import com.hazelcast.map.record.OffHeapRecordFactory;
import com.hazelcast.map.record.RecordFactory;
import com.hazelcast.map.writebehind.DelayedEntry;
import com.hazelcast.map.writebehind.WriteBehindManager;
import com.hazelcast.map.writebehind.WriteBehindManagers;
import com.hazelcast.map.writebehind.store.StoreEvent;
import com.hazelcast.map.writebehind.store.StoreListener;
import com.hazelcast.nio.ClassLoaderUtil;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.partition.InternalPartition;
import com.hazelcast.query.impl.IndexService;
import com.hazelcast.spi.NodeEngine;
import com.hazelcast.spi.Operation;
import com.hazelcast.spi.OperationService;
import com.hazelcast.util.ExceptionUtil;
import com.hazelcast.util.UuidUtil;
import com.hazelcast.wan.WanReplicationPublisher;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

/**
 * Map container.
 */
public class MapContainer {

    private volatile MapConfig mapConfig;
    private final String name;
    private final RecordFactory recordFactory;
    private final MapService mapService;
    private MapStoreWrapper storeWrapper;
    private final List<MapInterceptor> interceptors;
    private final Map<String, MapInterceptor> interceptorMap;
    private final IndexService indexService = new IndexService();
    private final boolean nearCacheEnabled;
    private final ReachabilityHandlerChain reachabilityHandlerChain;
    private final SizeEstimator nearCacheSizeEstimator;
    private final Map<Data, Object> initialKeys = new ConcurrentHashMap<Data, Object>();
    private final PartitioningStrategy partitioningStrategy;
    private WriteBehindManager writeBehindQueueManager;
    private WanReplicationPublisher wanReplicationPublisher;
    private MapMergePolicy wanMergePolicy;
    private final boolean evictionEnabled;

    public MapContainer(final String name, final MapConfig mapConfig, final MapService mapService) {
        this.name = name;
        this.mapConfig = mapConfig;
        this.mapService = mapService;
        this.partitioningStrategy = createPartitioningStrategy();
        this.reachabilityHandlerChain = ReachabilityHandlers.newHandlerChain(MapContainer.this);
        final NodeEngine nodeEngine = mapService.getNodeEngine();
        recordFactory = createRecordFactory(nodeEngine);
        initMapStoreOperations(nodeEngine);
        initWanReplication(nodeEngine);
        interceptors = new CopyOnWriteArrayList<MapInterceptor>();
        interceptorMap = new ConcurrentHashMap<String, MapInterceptor>();
        nearCacheEnabled = mapConfig.getNearCacheConfig() != null;
        nearCacheSizeEstimator = SizeEstimators.createNearCacheSizeEstimator();
        evictionEnabled = !MapConfig.EvictionPolicy.NONE.equals(mapConfig.getEvictionPolicy());
        mapService.getNodeEngine().getExecutionService()
                .scheduleAtFixedRate(new ClearExpiredRecordsTask(), 5, 5, TimeUnit.SECONDS);
    }

    /**
     * Periodically clears expired entries.(ttl & idle)
     */
    private class ClearExpiredRecordsTask implements Runnable {

        public void run() {
            final MapService service = mapService;
            final NodeEngine nodeEngine = service.getNodeEngine();
            for (int partitionId = 0; partitionId < nodeEngine.getPartitionService().getPartitionCount(); partitionId++) {
                InternalPartition partition = nodeEngine.getPartitionService().getPartition(partitionId);
                if (partition.isOwnerOrBackup(nodeEngine.getThisAddress())) {
                    final Operation expirationOperation = createExpirationOperation(partitionId);
                    OperationService operationService = mapService.getNodeEngine().getOperationService();
                    operationService.executeOperation(expirationOperation);
                }
            }
        }
    }

    private Operation createExpirationOperation(int partitionId) {
        final ClearExpiredOperation clearExpiredOperation = new ClearExpiredOperation(name);
        clearExpiredOperation
                .setNodeEngine(mapService.getNodeEngine())
                .setCallerUuid(mapService.getNodeEngine().getLocalMember().getUuid())
                .setPartitionId(partitionId)
                .setValidateTarget(false)
                .setService(mapService);
        return clearExpiredOperation;
    }

    public boolean isEvictionEnabled() {
        return evictionEnabled;
    }

    private void initMapStoreOperations(NodeEngine nodeEngine) {
        if (!isMapStoreEnabled()) {
            this.writeBehindQueueManager = WriteBehindManagers.emptyWriteBehindManager();
            return;
        }
        storeWrapper = createMapStoreWrapper(mapConfig.getMapStoreConfig(), nodeEngine);
        if (storeWrapper != null) {
            initMapStore(storeWrapper.getImpl(), mapConfig.getMapStoreConfig(), nodeEngine);
        }
        if (isWriteBehindMapStoreEnabled()) {
            this.writeBehindQueueManager
                    = WriteBehindManagers.createWriteBehindManager(name, mapService, storeWrapper);
            this.writeBehindQueueManager.addStoreListener(new StoreListener<DelayedEntry>() {
                @Override
                public void beforeStore(StoreEvent<DelayedEntry> storeEvent) {

                }

                @Override
                public void afterStore(StoreEvent<DelayedEntry> storeEvent) {
                    final DelayedEntry delayedEntry = storeEvent.getSource();
                    final Object value = delayedEntry.getValue();
                    // only process store delete operations.
                    if (value != null) {
                        return;
                    }
                    final Data key = (Data) storeEvent.getSource().getKey();
                    final int partitionId = mapService.getNodeEngine().getPartitionService().getPartitionId(key);
                    final PartitionContainer partitionContainer = mapService.getPartitionContainer(partitionId);
                    final RecordStore recordStore = partitionContainer.getExistingRecordStore(name);
                    if (recordStore != null) {
                        recordStore.removeFromWriteBehindWaitingDeletions(key);
                    }
                }
            });
            this.writeBehindQueueManager.start();

        }
    }

    private RecordFactory createRecordFactory(NodeEngine nodeEngine) {
        RecordFactory recordFactory;
        switch (mapConfig.getInMemoryFormat()) {
            case BINARY:
                recordFactory = new DataRecordFactory(mapConfig, nodeEngine.getSerializationService(), partitioningStrategy);
                break;
            case OBJECT:
                recordFactory = new ObjectRecordFactory(mapConfig, nodeEngine.getSerializationService());
                break;
            case OFFHEAP:
                recordFactory = new OffHeapRecordFactory(mapConfig, nodeEngine.getOffHeapStorage(),
                        nodeEngine.getSerializationService(), partitioningStrategy);
                break;
            default:
                throw new IllegalArgumentException("Invalid storage format: " + mapConfig.getInMemoryFormat());
        }
        return recordFactory;
    }

    public boolean isMapStoreEnabled() {<commit_msg>init write behind map store if enabled<commit_after>/*
 * Copyright (c) 2008-2013, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * 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 com.hazelcast.map;

import com.hazelcast.config.MapConfig;
import com.hazelcast.config.MapStoreConfig;
import com.hazelcast.config.PartitioningStrategyConfig;
import com.hazelcast.config.WanReplicationRef;
import com.hazelcast.core.MapLoaderLifecycleSupport;
import com.hazelcast.core.MapStoreFactory;
import com.hazelcast.core.PartitioningStrategy;
import com.hazelcast.map.eviction.ReachabilityHandlerChain;
import com.hazelcast.map.eviction.ReachabilityHandlers;
import com.hazelcast.map.merge.MapMergePolicy;
import com.hazelcast.map.operation.ClearExpiredOperation;
import com.hazelcast.map.record.DataRecordFactory;
import com.hazelcast.map.record.ObjectRecordFactory;
import com.hazelcast.map.record.OffHeapRecordFactory;
import com.hazelcast.map.record.RecordFactory;
import com.hazelcast.map.writebehind.DelayedEntry;
import com.hazelcast.map.writebehind.WriteBehindManager;
import com.hazelcast.map.writebehind.WriteBehindManagers;
import com.hazelcast.map.writebehind.store.StoreEvent;
import com.hazelcast.map.writebehind.store.StoreListener;
import com.hazelcast.nio.ClassLoaderUtil;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.partition.InternalPartition;
import com.hazelcast.query.impl.IndexService;
import com.hazelcast.spi.NodeEngine;
import com.hazelcast.spi.Operation;
import com.hazelcast.spi.OperationService;
import com.hazelcast.util.ExceptionUtil;
import com.hazelcast.util.UuidUtil;
import com.hazelcast.wan.WanReplicationPublisher;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

/**
 * Map container.
 */
public class MapContainer {

    private volatile MapConfig mapConfig;
    private final String name;
    private final RecordFactory recordFactory;
    private final MapService mapService;
    private MapStoreWrapper storeWrapper;
    private final List<MapInterceptor> interceptors;
    private final Map<String, MapInterceptor> interceptorMap;
    private final IndexService indexService = new IndexService();
    private final boolean nearCacheEnabled;
    private final ReachabilityHandlerChain reachabilityHandlerChain;
    private final SizeEstimator nearCacheSizeEstimator;
    private final Map<Data, Object> initialKeys = new ConcurrentHashMap<Data, Object>();
    private final PartitioningStrategy partitioningStrategy;
    private WriteBehindManager writeBehindQueueManager;
    private WanReplicationPublisher wanReplicationPublisher;
    private MapMergePolicy wanMergePolicy;
    private final boolean evictionEnabled;

    public MapContainer(final String name, final MapConfig mapConfig, final MapService mapService) {
        this.name = name;
        this.mapConfig = mapConfig;
        this.mapService = mapService;
        this.partitioningStrategy = createPartitioningStrategy();
        this.reachabilityHandlerChain = ReachabilityHandlers.newHandlerChain(MapContainer.this);
        final NodeEngine nodeEngine = mapService.getNodeEngine();
        recordFactory = createRecordFactory(nodeEngine);
        initMapStoreOperations(nodeEngine);
        initWanReplication(nodeEngine);
        interceptors = new CopyOnWriteArrayList<MapInterceptor>();
        interceptorMap = new ConcurrentHashMap<String, MapInterceptor>();
        nearCacheEnabled = mapConfig.getNearCacheConfig() != null;
        nearCacheSizeEstimator = SizeEstimators.createNearCacheSizeEstimator();
        evictionEnabled = !MapConfig.EvictionPolicy.NONE.equals(mapConfig.getEvictionPolicy());
        mapService.getNodeEngine().getExecutionService()
                .scheduleAtFixedRate(new ClearExpiredRecordsTask(), 5, 5, TimeUnit.SECONDS);
    }

    /**
     * Periodically clears expired entries.(ttl & idle)
     */
    private class ClearExpiredRecordsTask implements Runnable {

        public void run() {
            final MapService service = mapService;
            final NodeEngine nodeEngine = service.getNodeEngine();
            for (int partitionId = 0; partitionId < nodeEngine.getPartitionService().getPartitionCount(); partitionId++) {
                InternalPartition partition = nodeEngine.getPartitionService().getPartition(partitionId);
                if (partition.isOwnerOrBackup(nodeEngine.getThisAddress())) {
                    final Operation expirationOperation = createExpirationOperation(partitionId);
                    OperationService operationService = mapService.getNodeEngine().getOperationService();
                    operationService.executeOperation(expirationOperation);
                }
            }
        }
    }

    private Operation createExpirationOperation(int partitionId) {
        final ClearExpiredOperation clearExpiredOperation = new ClearExpiredOperation(name);
        clearExpiredOperation
                .setNodeEngine(mapService.getNodeEngine())
                .setCallerUuid(mapService.getNodeEngine().getLocalMember().getUuid())
                .setPartitionId(partitionId)
                .setValidateTarget(false)
                .setService(mapService);
        return clearExpiredOperation;
    }

    public boolean isEvictionEnabled() {
        return evictionEnabled;
    }

    private void initMapStoreOperations(NodeEngine nodeEngine) {
        if (!isMapStoreEnabled()) {
            this.writeBehindQueueManager = WriteBehindManagers.emptyWriteBehindManager();
            return;
        }
        storeWrapper = createMapStoreWrapper(mapConfig.getMapStoreConfig(), nodeEngine);
        if (storeWrapper != null) {
            initMapStore(storeWrapper.getImpl(), mapConfig.getMapStoreConfig(), nodeEngine);
        }
        if (!isWriteBehindMapStoreEnabled()) {
            this.writeBehindQueueManager = WriteBehindManagers.emptyWriteBehindManager();
            return;
        }
        initWriteBehindMapStore();
    }

    private void initWriteBehindMapStore() {
        if (!isWriteBehindMapStoreEnabled()) {
            return;
        }
        this.writeBehindQueueManager
                = WriteBehindManagers.createWriteBehindManager(name, mapService, storeWrapper);
        // listener for delete operations.
        this.writeBehindQueueManager.addStoreListener(new StoreListener<DelayedEntry>() {
            @Override
            public void beforeStore(StoreEvent<DelayedEntry> storeEvent) {

            }

            @Override
            public void afterStore(StoreEvent<DelayedEntry> storeEvent) {
                final DelayedEntry delayedEntry = storeEvent.getSource();
                final Object value = delayedEntry.getValue();
                // only process store delete operations.
                if (value != null) {
                    return;
                }
                final Data key = (Data) storeEvent.getSource().getKey();
                final int partitionId = mapService.getNodeEngine().getPartitionService().getPartitionId(key);
                final PartitionContainer partitionContainer = mapService.getPartitionContainer(partitionId);
                final RecordStore recordStore = partitionContainer.getExistingRecordStore(name);
                if (recordStore != null) {
                    recordStore.removeFromWriteBehindWaitingDeletions(key);
                }
            }
        });
        this.writeBehindQueueManager.start();
    }

    private RecordFactory createRecordFactory(NodeEngine nodeEngine) {
        RecordFactory recordFactory;
        switch (mapConfig.getInMemoryFormat()) {
            case BINARY:
                recordFactory = new DataRecordFactory(mapConfig, nodeEngine.getSerializationService(), partitioningStrategy);
                break;
            case OBJECT:
                recordFactory = new ObjectRecordFactory(mapConfig, nodeEngine.getSerializationService());
                break;
            case OFFHEAP:
                recordFactory = new OffHeapRecordFactory(mapConfig, nodeEngine.getOffHeapStorage(),
                        nodeEngine.getSerializationService(), partitioningStrategy);
                break;
            default:
                throw new IllegalArgumentException("Invalid storage format: " + mapConfig.getInMemoryFormat());
        }
        return recordFactory;
    }

    public boolean isMapStoreEnabled() {
{'fraction_non_alphanumeric': 0.07714766984839977, 'fraction_numerical': 0.0016844469399213925, 'mean_word_length': 3.2166193181818183, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '12735', 'n_tokens_mistral': 4537, 'n_tokens_neox': 4402, 'n_words': 954}