Month: March 2017

Recursively GREP for specific content

nohup  egrep -rnw '=\s112'  --include=*.java ./ 2>&1 | tee ~/112-audit-nick.txt

March 10, 2017

Data retrieval service with exponential backoff

Here we will create Data retrieval service with exponential backoff that we covered in the previous post.

Implementation


package com.rms.blueprint.data;

import java.util.Date;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.ObjLongConsumer;
import java.util.function.Supplier;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DataRetrievalWithBackoff implements Runnable
{
    public final Logger LOGGER = LoggerFactory.getLogger(DataRetrievalWithBackoff.class);

    private final Supplier pendingSupplier;

    private final ObjLongConsumer readyConsumer;

    private final Function capacitySupplier;

    private final long minLoadDurationInSeconds;

    private final long capacity;

    private final long maxBackoffDelayInSeconds;

    /**
     * @param config
     *            Configuration properties
     * @param pendingSupplier
     *            Supplier that tells us how many items is being processed at
     *            this time
     * @param readyConsumer
     *            Consumer that will be called when data is ready to load
     * @param capacitySupplier
     *            Function to calculate current capacity
     */
    private DataRetrievalWithBackoff(final long capacity, final long minLoadDurationInSeconds,
            final long maxBackoffDelayInSeconds, final Supplier pendingSupplier,
            final ObjLongConsumer readyConsumer, final Function capacitySupplier)
    {
        Objects.requireNonNull(pendingSupplier);
        Objects.requireNonNull(readyConsumer);
        Objects.requireNonNull(capacitySupplier);

        this.capacity = capacity;
        this.minLoadDurationInSeconds = minLoadDurationInSeconds;
        this.maxBackoffDelayInSeconds = maxBackoffDelayInSeconds;

        this.pendingSupplier = pendingSupplier;
        this.readyConsumer = readyConsumer;
        this.capacitySupplier = capacitySupplier;

        LOGGER.info(String.format("capacity ", capacity));
        LOGGER.info(String.format("minLoadDurationInSeconds ", minLoadDurationInSeconds));
        LOGGER.info(String.format("maxBackoffDelayInSeconds ", maxBackoffDelayInSeconds));

    }

    @Override
    public void run()
    {
        LOGGER.info("Running Data Retrieval");

        long lastLoadedTime = 0l;
        int attempt = 0;

        while (true)
        {
            if (Thread.currentThread().isInterrupted())
            {
                LOGGER.trace("Interrupted stopping [while]");
                break;
            }

            final long delta = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - lastLoadedTime);
            final long pending = pendingSupplier.get();

            final long backoffTime = DataUtil.backoff(attempt,
                                                      maxBackoffDelayInSeconds,
                                                      minLoadDurationInSeconds / 2.0);

            LOGGER.trace("Loading : lastLoaded : {} > {}  delta(s) {} pending : {} backoffTime  = {}",
                         new Object[] { lastLoadedTime, new Date(lastLoadedTime), delta, pending, backoffTime });

            if (delta >= minLoadDurationInSeconds && pending <= capacitySupplier.apply(capacity))
            {
                LOGGER.info("Loading : lastLoaded :  {} >  {}  delta(s) {} pending : {}",
                            new Object[] { lastLoadedTime, new Date(lastLoadedTime), delta, pending });

                // let the consumer know that we are ready
                readyConsumer.accept(backoffTime, attempt);

                if (pending == 0)
                    ++attempt;
                else
                    attempt = 0;

                lastLoadedTime = System.currentTimeMillis();
            }
            else
            {
                ++attempt;
            }

            try
            {

                Thread.sleep(TimeUnit.SECONDS.toMillis(backoffTime));
            }
            catch (final InterruptedException e)
            {
                LOGGER.trace("Interrupted stopping  [sleep]");
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    public static class Builder
    {
        private final Function DEFAULT_CAPACITY_SUPPLIER = (capacity) -> capacity / 2;

        private long minLoadDurationInSeconds = 60;

        private long capacity = 100;

        private long maxBackoffDelayInSeconds = 120;

        private Supplier pendingSupplier;

        private ObjLongConsumer readyConsumer;

        private Function capacitySupplier;

        public Builder capacity(final long capacity)
        {
            this.capacity = capacity;
            return this;
        }

        public Builder maxBackoffDelay(final long duration, final TimeUnit unit)
        {
            Objects.requireNonNull(unit);
            this.maxBackoffDelayInSeconds = unit.toSeconds(duration);
            return this;
        }

        public Builder minLoadDuration(final long duration, final TimeUnit unit)
        {
            Objects.requireNonNull(unit);
            this.minLoadDurationInSeconds = unit.toSeconds(duration);
            return this;
        }

        public Builder readyConsumer(final ObjLongConsumer readyConsumer)
        {
            Objects.requireNonNull(readyConsumer);
            this.readyConsumer = readyConsumer;
            return this;
        }

        public Builder capacitySupplier(final Function capacitySupplier)
        {
            Objects.requireNonNull(capacitySupplier);
            this.capacitySupplier = capacitySupplier;
            return this;
        }

        public Builder pendingSupplier(final Supplier pendingSupplier)
        {
            Objects.requireNonNull(capacitySupplier);
            this.pendingSupplier = pendingSupplier;
            return this;
        }

        public DataRetrievalWithBackoff build()
        {
            // check invariant
            Objects.requireNonNull(pendingSupplier, "Pening items supplier not provided");
            Objects.requireNonNull(readyConsumer, "Ready Consumer not provided");

            if (capacitySupplier == null)
                capacitySupplier = DEFAULT_CAPACITY_SUPPLIER;

            return new DataRetrievalWithBackoff(capacity,
                                                minLoadDurationInSeconds,
                                                maxBackoffDelayInSeconds,
                                                pendingSupplier,
                                                readyConsumer,
                                                capacitySupplier == null ? DEFAULT_CAPACITY_SUPPLIER
                                                    : capacitySupplier);
        }
    }
}

Usage


   //@formatter:off
            final DataRetrievalWithBackoff service = new DataRetrievalWithBackoff.Builder()
                    .capacity(1000)
                    .maxBackoffDelay(100, TimeUnit.SECONDS)
                    .minLoadDuration(10, TimeUnit.SECONDS)
                    .pendingSupplier(() -> getNumberOfPendingItemsToProcess())
                    .readyConsumer((time, attempt) -> fire(new DataLoadEvent()))
                .build(); 
           //@formatter:on

March 10, 2017

Exponential backoff

In a variety of computer networks, binary exponential backoff or truncated binary exponential backoff refers to an algorithm used to space out repeated retransmissions of the same block of data, often as part of network congestion avoidance.

Wikipedia Exponential backoff

Here is an implementation in Java

    /**
     * Calculate Exponential backoff
     * 
     * @param attempt
     *            number that we are checking
     * @param maxDelayInSeconds
     *            Max amount of time to wait
     * @param multiplier
     *            How much of backoff to perform
     * @return
     */
    public static long backoff(final int attempt, final long maxDelayInSeconds, final double multiplier)
    {
        final double delayInSec = (Math.pow(2.0, attempt) - 1.0) * .5;
        return Math.round(Math.min(delayInSec * multiplier, maxDelayInSeconds));
    }

Example

Here we have exponential backoff defined with three different parameters for the muliplier and 120 seconds as the max time.



System.out.println(String.format("Attempt\t\t 1\t4\t8\n"));

for (int i = 0; i < 10; i++)
{
    final long b1 = backoff(i, 120, 1);
    final long b2 = backoff(i, 120, 4);
    final long b3 = backoff(i, 120, 8);

    System.out.println(String.format("%d\t\t %d\t%d\t%d", i, b1, b2, b3));    
}


		 1	4	8

0		 0	0	0
1		 1	2	4
2		 2	6	12
3		 4	14	28
4		 8	30	60
5		 16	62	120
6		 32	120	120
7		 64	120	120
8		 120	120	120
9		 120	120	120


From the results above we can see that changing the multiplier can have significant implications. The larger the multiplier the faster we will be approaching our maxDelay and we will have longer paused between each attempt.
In next post, we will create a Data Retrieval Service that will utilize Exponential backoff.


			March 10, 2017