<commit_before>/*
 * Copyright 2013 Jive Software, Inc
 *
 * 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.jivesoftware.os.amza.service;

import com.google.common.base.Optional;
import com.google.common.base.Predicate;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.jivesoftware.os.amza.service.replication.AmzaRegionChangeReceiver;
import com.jivesoftware.os.amza.service.replication.AmzaRegionChangeReplicator;
import com.jivesoftware.os.amza.service.replication.AmzaRegionChangeTaker;
import com.jivesoftware.os.amza.service.replication.AmzaRegionCompactor;
import com.jivesoftware.os.amza.service.replication.RegionBackHighwaterMarks;
import com.jivesoftware.os.amza.service.replication.RegionStripe;
import com.jivesoftware.os.amza.service.replication.RegionStripeProvider;
import com.jivesoftware.os.amza.service.replication.SendFailureListener;
import com.jivesoftware.os.amza.service.replication.TakeFailureListener;
import com.jivesoftware.os.amza.service.storage.RegionIndex;
import com.jivesoftware.os.amza.service.storage.RegionPropertyMarshaller;
import com.jivesoftware.os.amza.service.storage.RegionProvider;
import com.jivesoftware.os.amza.service.storage.SystemStripeWALStorage;
import com.jivesoftware.os.amza.service.storage.WALs;
import com.jivesoftware.os.amza.service.storage.delta.DeltaStripeWALStorage;
import com.jivesoftware.os.amza.service.storage.delta.DeltaWALFactory;
import com.jivesoftware.os.amza.shared.RegionName;
import com.jivesoftware.os.amza.shared.RingHost;
import com.jivesoftware.os.amza.shared.RowChanges;
import com.jivesoftware.os.amza.shared.UpdatesSender;
import com.jivesoftware.os.amza.shared.UpdatesTaker;
import com.jivesoftware.os.amza.shared.WALStorageProvider;
import com.jivesoftware.os.amza.shared.stats.AmzaStats;
import com.jivesoftware.os.amza.storage.binary.BinaryRowIOProvider;
import com.jivesoftware.os.amza.storage.binary.BinaryRowMarshaller;
import com.jivesoftware.os.amza.storage.binary.RowIOProvider;
import com.jivesoftware.os.jive.utils.ordered.id.TimestampedOrderIdProvider;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class AmzaServiceInitializer {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    public static class AmzaServiceConfig {

        public String[] workingDirectories = new String[]{"./var/data/"};

        public int resendReplicasIntervalInMillis = 1000;
        public int applyReplicasIntervalInMillis = 1000;
        public int takeFromNeighborsIntervalInMillis = 1000;

        public long checkIfCompactionIsNeededIntervalInMillis = 60_000;
        public long compactTombstoneIfOlderThanNMillis = 30 * 24 * 60 * 60 * 1000L;

        public int numberOfResendThreads = 8;
        public int numberOfApplierThreads = 8;
        public int numberOfCompactorThreads = 8;
        public int numberOfTakerThreads = 8;
        public int numberOfReplicatorThreads = 24;

        public int corruptionParanoiaFactor = 10;

    }

    public AmzaService initialize(AmzaServiceConfig config,
        AmzaStats amzaStats,
        BinaryRowMarshaller rowMarshaller,
        RingHost ringHost,
        TimestampedOrderIdProvider orderIdProvider,
        RegionPropertyMarshaller regionPropertyMarshaller,
        WALStorageProvider regionsWALStorageProvider,
        WALStorageProvider replicaWALStorageProvider,
        WALStorageProvider resendWALStorageProvider,
        UpdatesSender updatesSender,
        UpdatesTaker updatesTaker,
        Optional<SendFailureListener> sendFailureListener,
        Optional<TakeFailureListener> takeFailureListener,
        RowChanges allRowChanges) throws Exception {

        AmzaRegionWatcher amzaRegionWatcher = new AmzaRegionWatcher(allRowChanges);

        RowIOProvider ioProvider = new BinaryRowIOProvider(amzaStats.ioStats, config.corruptionParanoiaFactor);

        RegionIndex regionIndex = new RegionIndex(amzaStats, config.workingDirectories, "amza/stores", regionsWALStorageProvider, regionPropertyMarshaller);
        regionIndex.open();

        AmzaRingReader amzaReadHostRing = new AmzaRingReader(ringHost, regionIndex);

        WALs resendWALs = new WALs(config.workingDirectories, "amza/WAL/resend", resendWALStorageProvider);
        resendWALs.load();

        AmzaRegionChangeReplicator replicator = new AmzaRegionChangeReplicator(amzaStats,
            rowMarshaller,
            amzaReadHostRing,
            regionIndex,
            resendWALs,
            updatesSender,
            Executors.newFixedThreadPool(config.numberOfReplicatorThreads),
            sendFailureListener,
            config.resendReplicasIntervalInMillis,
            config.numberOfResendThreads);

        RegionStripe systemRegionStripe = new RegionStripe(amzaStats, orderIdProvider, regionIndex, new SystemStripeWALStorage(), amzaRegionWatcher,
            new Predicate<RegionName>() {

                @Override
                public boolean apply(RegionName input) {
                    return input.isSystemRegion();
                }
            });

        final int deltaStorageStripes = 4;
        long maxUpdatesBeforeCompaction = 400_000;
        long compactAfterNUpdates = maxUpdatesBeforeCompaction / deltaStorageStripes;

        RegionStripe[] regionStripes = new RegionStripe[deltaStorageStripes];
        for (int i = 0; i < deltaStorageStripes; i++) {
            File walDir = new File(config.workingDirectories[i % config.workingDirectories.length], "delta-wal-" + i);
            DeltaWALFactory deltaWALFactory = new DeltaWALFactory(orderIdProvider, walDir, ioProvider, rowMarshaller);
            DeltaStripeWALStorage deltaWALStorage = new DeltaStripeWALStorage(i, rowMarshaller, deltaWALFactory, compactAfterNUpdates);
            final int stripeId = i;
            regionStripes[i] = new RegionStripe(amzaStats, orderIdProvider, regionIndex, deltaWALStorage, amzaRegionWatcher,
                new Predicate<RegionName>() {

                    @Override
                    public boolean apply(RegionName input) {
                        if (!input.isSystemRegion()) {

                            return Math.abs(input.hashCode()) % deltaStorageStripes == stripeId;
                        }
                        return false;
                    }
                });
        }

        RegionStripeProvider regionStripeProvider = new RegionStripeProvider(systemRegionStripe, regionIndex, regionStripes);

        RegionProvider regionProvider = new RegionProvider(
            orderIdProvider,
            regionPropertyMarshaller,
            replicator,
            regionIndex,
            allRowChanges);

        ExecutorService stripeLoaderThreadPool = Executors.newFixedThreadPool(regionStripes.length,
            new ThreadFactoryBuilder().setNameFormat("load-stripes-%d").build());
        List<Future> futures = new ArrayList<>();
        for (final RegionStripe regionStripe : regionStripes) {
            futures.add(stripeLoaderThreadPool.submit(new Runnable() {

                @Override
                public void run() {
                    try {
                        regionStripe.load();
                    } catch (Exception x) {
                        LOG.error("Failed while loading " + regionStripe, x);
                        throw new RuntimeException(x);
                    }
                }
            }));
        }
        for (Future future : futures) {
            future.get();
        }
        stripeLoaderThreadPool.shutdown();

        ScheduledExecutorService compactDeltasThreadPool = Executors.newScheduledThreadPool(config.numberOfCompactorThreads,
            new ThreadFactoryBuilder().setNameFormat("compact-deltas-%d").build());
        for (final RegionStripe regionStripe : regionStripes) {
            compactDeltasThreadPool.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        regionStripe.compact();
                    } catch (Throwable x) {
                        LOG.error("Compactor failed.", x);
                    }
                }
            }, 1, 1, TimeUnit.MINUTES); // TODO expose to config
        }

        AmzaHostRing amzaRing = new AmzaHostRing(amzaReadHostRing, systemRegionStripe, replicator, orderIdProvider);

        RegionBackHighwaterMarks highwaterMarks = new RegionBackHighwaterMarks(orderIdProvider, ringHost, systemRegionStripe, replicator);<commit_msg>Exposed striping options to config.<commit_after>/*
 * Copyright 2013 Jive Software, Inc
 *
 * 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.jivesoftware.os.amza.service;

import com.google.common.base.Optional;
import com.google.common.base.Predicate;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.jivesoftware.os.amza.service.replication.AmzaRegionChangeReceiver;
import com.jivesoftware.os.amza.service.replication.AmzaRegionChangeReplicator;
import com.jivesoftware.os.amza.service.replication.AmzaRegionChangeTaker;
import com.jivesoftware.os.amza.service.replication.AmzaRegionCompactor;
import com.jivesoftware.os.amza.service.replication.RegionBackHighwaterMarks;
import com.jivesoftware.os.amza.service.replication.RegionStripe;
import com.jivesoftware.os.amza.service.replication.RegionStripeProvider;
import com.jivesoftware.os.amza.service.replication.SendFailureListener;
import com.jivesoftware.os.amza.service.replication.TakeFailureListener;
import com.jivesoftware.os.amza.service.storage.RegionIndex;
import com.jivesoftware.os.amza.service.storage.RegionPropertyMarshaller;
import com.jivesoftware.os.amza.service.storage.RegionProvider;
import com.jivesoftware.os.amza.service.storage.SystemStripeWALStorage;
import com.jivesoftware.os.amza.service.storage.WALs;
import com.jivesoftware.os.amza.service.storage.delta.DeltaStripeWALStorage;
import com.jivesoftware.os.amza.service.storage.delta.DeltaWALFactory;
import com.jivesoftware.os.amza.shared.RegionName;
import com.jivesoftware.os.amza.shared.RingHost;
import com.jivesoftware.os.amza.shared.RowChanges;
import com.jivesoftware.os.amza.shared.UpdatesSender;
import com.jivesoftware.os.amza.shared.UpdatesTaker;
import com.jivesoftware.os.amza.shared.WALStorageProvider;
import com.jivesoftware.os.amza.shared.stats.AmzaStats;
import com.jivesoftware.os.amza.storage.binary.BinaryRowIOProvider;
import com.jivesoftware.os.amza.storage.binary.BinaryRowMarshaller;
import com.jivesoftware.os.amza.storage.binary.RowIOProvider;
import com.jivesoftware.os.jive.utils.ordered.id.TimestampedOrderIdProvider;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class AmzaServiceInitializer {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    public static class AmzaServiceConfig {

        public String[] workingDirectories = new String[]{"./var/data/"};

        public int resendReplicasIntervalInMillis = 1000;
        public int applyReplicasIntervalInMillis = 1000;
        public int takeFromNeighborsIntervalInMillis = 1000;

        public long checkIfCompactionIsNeededIntervalInMillis = 60_000;
        public long compactTombstoneIfOlderThanNMillis = 30 * 24 * 60 * 60 * 1000L;

        public int numberOfResendThreads = 8;
        public int numberOfApplierThreads = 8;
        public int numberOfCompactorThreads = 8;
        public int numberOfTakerThreads = 8;
        public int numberOfReplicatorThreads = 24;

        public int corruptionParanoiaFactor = 10;

        public int numberOfDeltaStripes = 4;
        public int maxUpdatesBeforeDeltaStripeCompaction = 1_000_000;
        public int deltaStripeCompactionIntervalInMillis = 1_000 * 60;

    }

    public AmzaService initialize(AmzaServiceConfig config,
        AmzaStats amzaStats,
        BinaryRowMarshaller rowMarshaller,
        RingHost ringHost,
        TimestampedOrderIdProvider orderIdProvider,
        RegionPropertyMarshaller regionPropertyMarshaller,
        WALStorageProvider regionsWALStorageProvider,
        WALStorageProvider replicaWALStorageProvider,
        WALStorageProvider resendWALStorageProvider,
        UpdatesSender updatesSender,
        UpdatesTaker updatesTaker,
        Optional<SendFailureListener> sendFailureListener,
        Optional<TakeFailureListener> takeFailureListener,
        RowChanges allRowChanges) throws Exception {

        AmzaRegionWatcher amzaRegionWatcher = new AmzaRegionWatcher(allRowChanges);

        RowIOProvider ioProvider = new BinaryRowIOProvider(amzaStats.ioStats, config.corruptionParanoiaFactor);

        RegionIndex regionIndex = new RegionIndex(amzaStats, config.workingDirectories, "amza/stores", regionsWALStorageProvider, regionPropertyMarshaller);
        regionIndex.open();

        AmzaRingReader amzaReadHostRing = new AmzaRingReader(ringHost, regionIndex);

        WALs resendWALs = new WALs(config.workingDirectories, "amza/WAL/resend", resendWALStorageProvider);
        resendWALs.load();

        AmzaRegionChangeReplicator replicator = new AmzaRegionChangeReplicator(amzaStats,
            rowMarshaller,
            amzaReadHostRing,
            regionIndex,
            resendWALs,
            updatesSender,
            Executors.newFixedThreadPool(config.numberOfReplicatorThreads),
            sendFailureListener,
            config.resendReplicasIntervalInMillis,
            config.numberOfResendThreads);

        RegionStripe systemRegionStripe = new RegionStripe(amzaStats, orderIdProvider, regionIndex, new SystemStripeWALStorage(), amzaRegionWatcher,
            new Predicate<RegionName>() {

                @Override
                public boolean apply(RegionName input) {
                    return input.isSystemRegion();
                }
            });

        final int deltaStorageStripes = config.numberOfDeltaStripes;
        long maxUpdatesBeforeCompaction = config.maxUpdatesBeforeDeltaStripeCompaction;
        
        RegionStripe[] regionStripes = new RegionStripe[deltaStorageStripes];
        for (int i = 0; i < deltaStorageStripes; i++) {
            File walDir = new File(config.workingDirectories[i % config.workingDirectories.length], "delta-wal-" + i);
            DeltaWALFactory deltaWALFactory = new DeltaWALFactory(orderIdProvider, walDir, ioProvider, rowMarshaller);
            DeltaStripeWALStorage deltaWALStorage = new DeltaStripeWALStorage(i, rowMarshaller, deltaWALFactory, maxUpdatesBeforeCompaction);
            final int stripeId = i;
            regionStripes[i] = new RegionStripe(amzaStats, orderIdProvider, regionIndex, deltaWALStorage, amzaRegionWatcher,
                new Predicate<RegionName>() {

                    @Override
                    public boolean apply(RegionName input) {
                        if (!input.isSystemRegion()) {

                            return Math.abs(input.hashCode()) % deltaStorageStripes == stripeId;
                        }
                        return false;
                    }
                });
        }

        RegionStripeProvider regionStripeProvider = new RegionStripeProvider(systemRegionStripe, regionIndex, regionStripes);

        RegionProvider regionProvider = new RegionProvider(
            orderIdProvider,
            regionPropertyMarshaller,
            replicator,
            regionIndex,
            allRowChanges);

        ExecutorService stripeLoaderThreadPool = Executors.newFixedThreadPool(regionStripes.length,
            new ThreadFactoryBuilder().setNameFormat("load-stripes-%d").build());
        List<Future> futures = new ArrayList<>();
        for (final RegionStripe regionStripe : regionStripes) {
            futures.add(stripeLoaderThreadPool.submit(new Runnable() {

                @Override
                public void run() {
                    try {
                        regionStripe.load();
                    } catch (Exception x) {
                        LOG.error("Failed while loading " + regionStripe, x);
                        throw new RuntimeException(x);
                    }
                }
            }));
        }
        for (Future future : futures) {
            future.get();
        }
        stripeLoaderThreadPool.shutdown();

        ScheduledExecutorService compactDeltasThreadPool = Executors.newScheduledThreadPool(config.numberOfCompactorThreads,
            new ThreadFactoryBuilder().setNameFormat("compact-deltas-%d").build());
        for (final RegionStripe regionStripe : regionStripes) {
            compactDeltasThreadPool.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        regionStripe.compact();
                    } catch (Throwable x) {
                        LOG.error("Compactor failed.", x);
                    }
                }
            }, config.deltaStripeCompactionIntervalInMillis, config.deltaStripeCompactionIntervalInMillis, TimeUnit.MILLISECONDS);
        }

        AmzaHostRing amzaRing = new AmzaHostRing(amzaReadHostRing, systemRegionStripe, replicator, orderIdProvider);

        RegionBackHighwaterMarks highwaterMarks = new RegionBackHighwaterMarks(orderIdProvider, ringHost, systemRegionStripe, replicator);
{'fraction_non_alphanumeric': 0.0764891934633632, 'fraction_numerical': 0.006062203479177649, 'mean_word_length': 3.0991789109766636, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 15, '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': '5331530', 'n_tokens_mistral': 5395, 'n_tokens_neox': 5168, 'n_words': 994}