You are viewing a plain text version of this content. The canonical link for it is here.

Posted to dev@apex.apache.org by chandnisingh <gi...@git.apache.org> on 2015/12/22 03:25:56 UTC

[GitHub] incubator-apex-malhar pull request: [For Review Only] MLHR-1897 #c...

GitHub user chandnisingh opened a pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145

    [For Review Only] MLHR-1897 #comment added managed state 

    Still working on tests and will add documentation. 
    
    @amberarrow @ilooner @PramodSSImmaneni 
    Please review

You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/chandnisingh/incubator-apex-malhar MLHR-1897

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/incubator-apex-malhar/pull/145.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

    This closes #145
    
----
commit 1ef879b55437c343d9fce414bd9d3c41db2da154
Author: Chandni Singh <cs...@apache.org>
Date:   2015-12-13T11:13:08Z

    MLHR-1897 #comment added managed state

----


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: [For Review Only] MLHR-1897 #c...

Posted by tweise <gi...@git.apache.org>.

Github user tweise commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r48512994
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/AbstractManagedStateImpl.java ---
    @@ -0,0 +1,487 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.io.IOException;
    +import java.lang.reflect.Array;
    +import java.util.Comparator;
    +import java.util.Map;
    +import java.util.Timer;
    +import java.util.TimerTask;
    +import java.util.concurrent.Callable;
    +import java.util.concurrent.ExecutorService;
    +import java.util.concurrent.Executors;
    +import java.util.concurrent.Future;
    +import java.util.concurrent.atomic.AtomicReference;
    +
    +import javax.validation.constraints.Min;
    +import javax.validation.constraints.NotNull;
    +
    +import org.joda.time.Duration;
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import com.esotericsoftware.kryo.serializers.FieldSerializer;
    +import com.esotericsoftware.kryo.serializers.JavaSerializer;
    +import com.google.common.base.Preconditions;
    +import com.google.common.collect.Maps;
    +import com.google.common.collect.MinMaxPriorityQueue;
    +import com.google.common.util.concurrent.Futures;
    +
    +import com.datatorrent.api.Component;
    +import com.datatorrent.api.Context.DAGContext;
    +import com.datatorrent.api.Context.OperatorContext;
    +import com.datatorrent.api.Operator;
    +import com.datatorrent.api.annotation.Stateless;
    +import com.datatorrent.common.util.NameableThreadFactory;
    +import com.datatorrent.lib.fileaccess.FileAccess;
    +import com.datatorrent.lib.fileaccess.TFileImpl;
    +import com.datatorrent.lib.util.comparator.SliceComparator;
    +import com.datatorrent.netlet.util.DTThrowable;
    +import com.datatorrent.netlet.util.Slice;
    +
    +/**
    + * Not thread safe.
    + */
    +public abstract class AbstractManagedStateImpl
    --- End diff --
    
    Javadoc?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1897 added managed ...

Posted by chandnisingh <gi...@git.apache.org>.

Github user chandnisingh commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r56759046
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/Bucket.java ---
    @@ -0,0 +1,534 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.io.IOException;
    +import java.util.Iterator;
    +import java.util.Map;
    +import java.util.Objects;
    +import java.util.Set;
    +import java.util.TreeMap;
    +import java.util.TreeSet;
    +import java.util.concurrent.atomic.AtomicLong;
    +
    +import javax.validation.constraints.NotNull;
    +
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import com.google.common.annotations.VisibleForTesting;
    +import com.google.common.base.Preconditions;
    +import com.google.common.collect.Maps;
    +import com.google.common.collect.Sets;
    +
    +import com.datatorrent.lib.fileaccess.FileAccess;
    +import com.datatorrent.netlet.util.Slice;
    +
    +/**
    + * A bucket that groups events.
    + */
    +public interface Bucket extends ManagedStateComponent
    +{
    +  /**
    +   * @return bucket id
    +   */
    +  long getBucketId();
    +
    +  /**
    +   *
    +   * @return size of bucket in memory.
    +   */
    +  long getSizeInBytes();
    +
    +  /**
    +   * Get value of a key.
    +   *
    +   * @param key        key.
    +   * @param timeBucket time bucket of the key if known; -1 otherwise.
    +   * @param source     source to read from
    +   * @return value of the key.
    +   */
    +  Slice get(Slice key, long timeBucket, ReadSource source);
    +
    +  /**
    +   * Set value of a key.
    +   *
    +   * @param key        key.
    +   * @param timeBucket timeBucket of the key.
    +   * @param value      value of the key.
    +   */
    +  void put(Slice key, long timeBucket, Slice value);
    +
    +  /**
    +   * Triggers the bucket to checkpoint. Returns the non checkpointed data so far.
    +   *
    +   * @return non checkpointed data.
    +   */
    +  Map<Slice, BucketedValue> checkpoint(long windowId);
    +
    +  /**
    +   * Triggers the bucket to commit data till provided window id.
    +   *
    +   * @param windowId window id
    +   */
    +  void committed(long windowId);
    +
    +  /**
    +   * Triggers bucket to free memory which is already persisted in bucket data files.
    +   *
    +   * @return amount of memory freed in bytes.
    +   * @throws IOException
    +   */
    +  long freeMemory() throws IOException;
    +
    +  /**
    +   * Allows the bucket to process/cache data which is recovered (from window files) after failure.
    +   *
    +   * @param windowId largest recovery window
    +   * @param recoveredData recovered data
    +   */
    +  void recoveredData(long windowId, Map<Slice, Bucket.BucketedValue> recoveredData);
    +
    +  enum ReadSource
    +  {
    +    MEMORY,      //state in memory in key/value form
    +    READERS,     //these are streams in which the key will be searched and serialized.
    +    ALL          //both the above states.
    +  }
    +
    +  class BucketedValue
    +  {
    +    private long timeBucket;
    +    private Slice value;
    +
    +    protected BucketedValue()
    +    {
    +    }
    +
    +    protected BucketedValue(long timeBucket, Slice value)
    +    {
    +      this.timeBucket = timeBucket;
    +      this.value = value;
    +    }
    +
    +    protected long getTimeBucket()
    +    {
    +      return timeBucket;
    +    }
    +
    +    protected void setTimeBucket(long timeBucket)
    +    {
    +      this.timeBucket = timeBucket;
    +    }
    +
    +    public Slice getValue()
    +    {
    +      return value;
    +    }
    +
    +    public void setValue(Slice value)
    +    {
    +      this.value = value;
    +    }
    +
    +    @Override
    +    public boolean equals(Object o)
    +    {
    +      if (this == o) {
    +        return true;
    +      }
    +      if (!(o instanceof BucketedValue)) {
    +        return false;
    +      }
    +
    +      BucketedValue that = (BucketedValue)o;
    +
    +      return timeBucket == that.timeBucket && value.equals(that.value);
    +
    +    }
    +
    +    @Override
    +    public int hashCode()
    +    {
    +      return Objects.hash(timeBucket, value);
    +    }
    +  }
    +
    +  /**
    +   * Default bucket.<br/>
    +   * Not thread-safe.
    +   */
    +  class DefaultBucket implements Bucket
    +  {
    +    private final long bucketId;
    +
    +    //Key -> Ordered values
    +    private Map<Slice, BucketedValue> flash = Maps.newHashMap();
    +
    +    //Data persisted in write ahead logs. window -> bucket
    +    private final transient TreeMap<Long, Map<Slice, BucketedValue>> checkpointedData = Maps.newTreeMap();
    +
    +    //Data persisted in bucket data files
    +    private final transient Map<Slice, BucketedValue> committedData = Maps.newHashMap();
    +
    +    //Data recovered
    +    private final transient TreeMap<Long, Map<Slice, BucketedValue>> recoveredData = Maps.newTreeMap();
    +
    +    //Data serialized/deserialized from bucket data files
    +    private final transient Map<Slice, BucketedValue> readCache = Maps.newConcurrentMap();
    +
    +    //TimeBucket -> FileReaders
    +    private final transient Map<Long, FileAccess.FileReader> readers = Maps.newTreeMap();
    +
    +    protected transient ManagedStateContext managedStateContext;
    +
    +    private AtomicLong sizeInBytes = new AtomicLong(0);
    +
    +    private final transient Slice dummyGetKey = new Slice(null, 0, 0);
    +
    +    private transient TreeSet<BucketsFileSystem.ImmutableTimeBucketMeta> cachedBucketMetas;
    +
    +    private DefaultBucket()
    +    {
    +      //for kryo
    +      bucketId = -1;
    +    }
    +
    +    protected DefaultBucket(long bucketId)
    +    {
    +      this.bucketId = bucketId;
    +    }
    +
    +    @Override
    +    public void setup(@NotNull ManagedStateContext managedStateContext)
    +    {
    +      this.managedStateContext = Preconditions.checkNotNull(managedStateContext, "managed state context");
    +    }
    +
    +    @Override
    +    public long getBucketId()
    +    {
    +      return bucketId;
    +    }
    +
    +    @Override
    +    public long getSizeInBytes()
    +    {
    +      return sizeInBytes.longValue();
    +    }
    +
    +    private Slice getFromMemory(Slice key)
    +    {
    +      //search the cache for key
    +      BucketedValue bucketedValue = flash.get(key);
    +      if (bucketedValue != null) {
    +        return bucketedValue.getValue();
    +      }
    +
    +      for (Long window : checkpointedData.descendingKeySet()) {
    +        //traverse the checkpointed data in reverse order
    +        bucketedValue = checkpointedData.get(window).get(key);
    +        if (bucketedValue != null) {
    +          return bucketedValue.getValue();
    +        }
    +      }
    +
    +      for (Long window : recoveredData.descendingKeySet()) {
    +        //traverse the reccovered data in reverse order
    +        bucketedValue = recoveredData.get(window).get(key);
    +        if (bucketedValue != null) {
    +          return bucketedValue.getValue();
    +        }
    +      }
    +
    +      bucketedValue = committedData.get(key);
    +      if (bucketedValue != null) {
    +        return bucketedValue.getValue();
    +      }
    +
    +      bucketedValue = readCache.get(key);
    +      if (bucketedValue != null) {
    +        return bucketedValue.getValue();
    +      }
    +      return null;
    +    }
    +
    +    private Slice getFromReaders(Slice key, long timeBucket)
    +    {
    +      if (timeBucket != -1) {
    +        Slice valSlice = getKeyFromTimeBucketReader(key, timeBucket);
    +        if (valSlice != null) {
    +          BucketedValue bucketedValue = new BucketedValue(timeBucket, valSlice);
    +          readCache.put(key, bucketedValue);
    --- End diff --
    
    Will fix this. However the expectation here is that the latest value is retrieved always. If the latest time-bucket is known then the process to read it is faster otherwise reading all time-buckets is slower. This is again missing from the documentation  and I will add to it.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1897 added managed ...

Posted by amberarrow <gi...@git.apache.org>.

Github user amberarrow commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r56766113
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/ManagedTimeUnifiedStateImpl.java ---
    @@ -0,0 +1,238 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.io.DataInputStream;
    +import java.io.DataOutputStream;
    +import java.io.IOException;
    +import java.util.concurrent.Future;
    +import java.util.concurrent.LinkedBlockingQueue;
    +
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import org.apache.hadoop.fs.LocatedFileStatus;
    +import org.apache.hadoop.fs.RemoteIterator;
    +
    +import com.google.common.base.Preconditions;
    +import com.google.common.collect.Queues;
    +import com.google.common.util.concurrent.Futures;
    +
    +import com.datatorrent.lib.fileaccess.FileAccess;
    +import com.datatorrent.lib.state.BucketedState;
    +import com.datatorrent.netlet.util.Slice;
    +
    +/**
    + * In this implementation of {@link ManagedState} the buckets in memory are time-buckets.
    + * <p/>
    + *
    + * <b>Difference from {@link ManagedTimeStateImpl}</b>: <br/>
    + * <ol>
    + * <li>The main buckets in {@link ManagedTimeStateImpl} are unique adhoc long ids which the user provides with the
    + * key. In this implementation the main buckets are time buckets. The user provides just the time and the time bucket is
    + * derived from it.
    + * </li>
    + * <br/>
    + *
    + * <li>In regards to the bucket data on disk, in {@link ManagedTimeStateImpl} the buckets are persisted on disk
    + * with each bucket data further grouped into time-buckets: {base_path}/{bucketId}/{time-bucket id}. <br/>
    + * In this implementation operator id is used as bucketId (on disk) and there is just one time-bucket under a
    + * particular operator id:
    + * {base_path}/{operator id}/{time bucket id}.
    + * </li>
    + * <br/>
    + *
    + * <li>In {@link ManagedTimeStateImpl} a bucket belongs to just one partition. Multiple partitions cannot write to
    + * the same bucket. <br/>
    + * In this implementation multiple partitions can be working with the same time-bucket (since time-bucket is derived
    + * from time). This works because on the disk the time-bucket data is segregated under each operator id.
    + * </li>
    + * <br/>
    + *
    + * <li>While {@link ManagedTimeStateImpl} can support dynamic partitioning by pre-allocating buckets this will not
    + * be able to support dynamic partitioning efficiently.
    + * </li>
    +
    + * </ol>
    + */
    +public class ManagedTimeUnifiedStateImpl extends AbstractManagedStateImpl implements BucketedState
    +{
    +  private final transient LinkedBlockingQueue<Long> purgedTimeBuckets = Queues.newLinkedBlockingQueue();
    +
    +  public ManagedTimeUnifiedStateImpl()
    +  {
    +    bucketsFileSystem = new TimeUnifiedBucketsFileSystem();
    +  }
    +
    +  @Override
    +  public int getNumBuckets()
    +  {
    +    return timeBucketAssigner.getNumBuckets();
    +  }
    +
    +  @Override
    +  public void put(long time, Slice key, Slice value)
    +  {
    +    long timeBucket = timeBucketAssigner.getTimeBucketFor(time);
    +    if (timeBucket == -1) {
    +      //time is expired so return null.
    +      return;
    +    }
    +    int bucketIdx = prepareBucket(timeBucket);
    +
    +    buckets[bucketIdx].put(key, timeBucket, value);
    +
    +  }
    +
    +  @Override
    +  public Slice getSync(long time, Slice key)
    +  {
    +    long timeBucket = timeBucketAssigner.getTimeBucketFor(time);
    +    if (timeBucket == -1) {
    +      //time is expired so return null.
    --- End diff --
    
    return null. => return expired token.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: MLHR-1897 #comment added manag...

Posted by chandnisingh <gi...@git.apache.org>.

Github user chandnisingh commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r53725996
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/Bucket.java ---
    @@ -0,0 +1,505 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.io.IOException;
    +import java.util.Iterator;
    +import java.util.Map;
    +import java.util.Objects;
    +import java.util.Set;
    +import java.util.TreeMap;
    +import java.util.TreeSet;
    +import java.util.concurrent.atomic.AtomicLong;
    +
    +import javax.validation.constraints.NotNull;
    +
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import com.google.common.annotations.VisibleForTesting;
    +import com.google.common.base.Preconditions;
    +import com.google.common.collect.Maps;
    +import com.google.common.collect.Sets;
    +
    +import com.datatorrent.api.Component;
    +import com.datatorrent.api.Context;
    +import com.datatorrent.lib.fileaccess.FileAccess;
    +import com.datatorrent.netlet.util.Slice;
    +
    +/**
    + * A bucket that groups events.
    + */
    +public interface Bucket extends Component<Context.OperatorContext>
    +{
    +  /**
    +   * @return bucket id
    +   */
    +  long getBucketId();
    +
    +  /**
    +   *
    +   * @return size of bucket in memory.
    +   */
    +  long getSizeInBytes();
    +
    +  /**
    +   * Get value of a key.
    +   *
    +   * @param key        key.
    +   * @param timeBucket time bucket of the key if known; -1 otherwise.
    +   * @param source     source to read from
    +   * @return value of the key.
    +   */
    +  Slice get(Slice key, long timeBucket, ReadSource source);
    +
    +  /**
    +   * Set value of a key.
    +   *
    +   * @param key        key.
    +   * @param timeBucket timeBucket of the key.
    +   * @param value      value of the key.
    +   */
    +  void put(Slice key, long timeBucket, Slice value);
    +
    +  /**
    +   * Triggers the bucket to checkpoint. Returns the non checkpointed data so far.
    +   *
    +   * @return non checkpointed data.
    +   */
    +  Map<Slice, BucketedValue> checkpoint(long windowId);
    +
    +  /**
    +   * Triggers the bucket to commit data till provided window id.
    +   *
    +   * @param windowId window id
    +   */
    +  void committed(long windowId);
    +
    +  /**
    +   * Triggers bucket to free memory which is already persisted in bucket data files.
    +   *
    +   * @return amount of memory freed in bytes.
    +   * @throws IOException
    +   */
    +  long freeMemory() throws IOException;
    +
    +  enum ReadSource
    +  {
    +    MEMORY,      //state in memory in key/value form
    +    READERS,     //these are streams in which the key will be searched and serialized.
    +    ALL          //both the above states.
    +  }
    +
    +  class BucketedValue
    +  {
    +    private long timeBucket;
    +    private Slice value;
    +
    +    protected BucketedValue()
    +    {
    +    }
    +
    +    protected BucketedValue(long timeBucket, Slice value)
    +    {
    +      this.timeBucket = timeBucket;
    +      this.value = value;
    +    }
    +
    +    protected long getTimeBucket()
    +    {
    +      return timeBucket;
    +    }
    +
    +    protected void setTimeBucket(long timeBucket)
    +    {
    +      this.timeBucket = timeBucket;
    +    }
    +
    +    public Slice getValue()
    +    {
    +      return value;
    +    }
    +
    +    public void setValue(Slice value)
    +    {
    +      this.value = value;
    +    }
    +
    +    @Override
    +    public boolean equals(Object o)
    +    {
    +      if (this == o) {
    +        return true;
    +      }
    +      if (!(o instanceof BucketedValue)) {
    +        return false;
    +      }
    +
    +      BucketedValue that = (BucketedValue)o;
    +
    +      return timeBucket == that.timeBucket && value.equals(that.value);
    +
    +    }
    +
    +    @Override
    +    public int hashCode()
    +    {
    +      return Objects.hash(timeBucket, value);
    +    }
    +  }
    +
    +  /**
    +   * Default bucket.<br/>
    +   * Not thread-safe.
    +   */
    +  class DefaultBucket implements Bucket
    +  {
    +    private final long bucketId;
    +
    +    //Key -> Ordered values
    +    private Map<Slice, BucketedValue> flash = Maps.newHashMap();
    +
    +    //Data persisted in write ahead logs. window -> bucket
    +    private final transient TreeMap<Long, Map<Slice, BucketedValue>> checkpointedData = Maps.newTreeMap();
    +
    +    //Data persisted in bucket data files
    +    private final transient Map<Slice, BucketedValue> committedData = Maps.newHashMap();
    +
    +    //Data serialized/deserialized from bucket data files
    +    private final transient Map<Slice, BucketedValue> readCache = Maps.newConcurrentMap();
    +
    +    //TimeBucket -> FileReaders
    +    private final transient Map<Long, FileAccess.FileReader> readers = Maps.newTreeMap();
    +
    +    private final transient ManagedStateContext managedStateContext;
    +
    +    private AtomicLong sizeInBytes = new AtomicLong(0);
    +
    +    private final transient Slice dummyGetKey = new Slice(null, 0, 0);
    +
    +    private transient TreeSet<BucketsMetaDataManager.ImmutableTimeBucketMeta> cachedBucketMetas;
    +
    +    private DefaultBucket()
    +    {
    +      //for kryo
    +      bucketId = -1;
    +      managedStateContext = null;
    +    }
    +
    +    protected DefaultBucket(long bucketId, @NotNull ManagedStateContext managedStateContext)
    +    {
    +      this.bucketId = bucketId;
    +      this.managedStateContext = Preconditions.checkNotNull(managedStateContext);
    +    }
    +
    +    @Override
    +    public void setup(Context.OperatorContext context)
    +    {
    +    }
    +
    +    @Override
    +    public long getBucketId()
    +    {
    +      return bucketId;
    +    }
    +
    +    @Override
    +    public long getSizeInBytes()
    +    {
    +      return sizeInBytes.longValue();
    +    }
    +
    +    private Slice getFromMemory(Slice key)
    +    {
    +      //search the cache for key
    +      BucketedValue bucketedValue = flash.get(key);
    +      if (bucketedValue != null) {
    +        return bucketedValue.getValue();
    +      }
    +
    +      for (Long window : checkpointedData.descendingKeySet()) {
    +        //traversing the checkpointed data in reverse order
    +        bucketedValue = checkpointedData.get(window).get(key);
    +        if (bucketedValue != null) {
    +          return bucketedValue.getValue();
    +        }
    +      }
    +
    +      bucketedValue = committedData.get(key);
    +      if (bucketedValue != null) {
    +        return bucketedValue.getValue();
    +      }
    +
    +      bucketedValue = readCache.get(key);
    +      if (bucketedValue != null) {
    +        return bucketedValue.getValue();
    +      }
    +      return null;
    +    }
    +
    +    private Slice getFromReaders(Slice key, long timeBucket)
    +    {
    +      if (timeBucket != -1) {
    +        Slice valSlice = getKeyFromTimeBucketReader(key, timeBucket);
    +        if (valSlice != null) {
    +          BucketedValue bucketedValue = new BucketedValue(timeBucket, valSlice);
    +          readCache.put(key, bucketedValue);
    +        }
    +        return valSlice;
    +      } else {
    +        //search all the time buckets
    +        try {
    +          if (cachedBucketMetas == null) {
    +            cachedBucketMetas = managedStateContext.getBucketsMetaDataManager().getAllTimeBuckets(bucketId);
    +          }
    +
    +          for (BucketsMetaDataManager.ImmutableTimeBucketMeta immutableTimeBucketMeta : cachedBucketMetas) {
    +
    +            if (managedStateContext.getKeyComparator().compare(key, immutableTimeBucketMeta.getFirstKey()) >= 0) {
    +              //keys in the time bucket files are sorted so if the first key in the file is greater than the key being
    +              //searched, the key will not be present in that file.
    +              Slice valSlice = getKeyFromTimeBucketReader(key, immutableTimeBucketMeta.getTimeBucketId());
    +              if (valSlice != null) {
    +                BucketedValue bucketedValue = new BucketedValue(immutableTimeBucketMeta.getTimeBucketId(), valSlice);
    +                readCache.put(key, bucketedValue);
    +                return valSlice;
    +              }
    +            }
    +          }
    +
    +        } catch (IOException e) {
    +          throw new RuntimeException("get time-buckets " + bucketId, e);
    +        }
    +      }
    +      return null;
    +    }
    +
    +    @Override
    +    public Slice get(Slice key, long timeBucket, ReadSource readSource)
    +    {
    +      switch (readSource) {
    +        case MEMORY:
    +          return getFromMemory(key);
    +        case READERS:
    +          return getFromReaders(key, timeBucket);
    +        case ALL:
    +        default:
    +          Slice value = getFromMemory(key);
    +          if (value != null) {
    +            return value;
    +          }
    +          return getFromReaders(key, timeBucket);
    +      }
    +    }
    +
    +    /**
    +     * Returns the value for the key from a time-bucket reader
    +     * @param key        key
    +     * @param timeBucket time bucket
    +     * @return value if key is found in the time bucket; false otherwise
    +     */
    +    private Slice getKeyFromTimeBucketReader(Slice key, long timeBucket)
    +    {
    +      FileAccess.FileReader fileReader = readers.get(timeBucket);
    +      if (fileReader != null) {
    +        return readKey(fileReader, key, timeBucket);
    +      }
    +      //file reader is not loaded and is null
    +      try {
    +        if (loadFileReader(timeBucket)) {
    +          return readKey(readers.get(timeBucket), key, timeBucket);
    +        }
    +        return null;
    +      } catch (IOException e) {
    +        throw new RuntimeException("while loading " + bucketId + ", " + timeBucket, e);
    +      }
    +    }
    +
    +    private Slice readKey(FileAccess.FileReader fileReader, Slice key, long timeBucket)
    +    {
    +      Slice valSlice = new Slice(null, 0, 0);
    +      try {
    +        if (fileReader.seek(key)) {
    +          fileReader.next(dummyGetKey, valSlice);
    +          return valSlice;
    +        } else {
    +          return null;
    +        }
    +      } catch (IOException e) {
    +        throw new RuntimeException("reading " + bucketId + ", " + timeBucket, e);
    +      }
    +    }
    +
    +    private boolean loadFileReader(long timeBucketId) throws IOException
    +    {
    +      BucketsMetaDataManager.ImmutableTimeBucketMeta tbm = managedStateContext.getBucketsMetaDataManager()
    +          .getTimeBucketMeta(bucketId, timeBucketId);
    +
    +      if (tbm != null) {
    +        FileAccess.FileReader reader = managedStateContext.getFileAccess().getReader(bucketId,
    +            BucketsDataManager.getFileName(timeBucketId));
    +        readers.put(timeBucketId, reader);
    +        sizeInBytes.getAndAdd(tbm.getSizeInBytes());
    +        return true;
    +      }
    +      return false;
    +    }
    +
    +    @Override
    +    public void put(Slice key, long timeBucket, Slice value)
    +    {
    +      BucketedValue bucketedValue = flash.get(key);
    +      if (bucketedValue == null) {
    +        bucketedValue = new BucketedValue();
    +        flash.put(key, bucketedValue);
    +        sizeInBytes.getAndAdd(key.length);
    +        sizeInBytes.getAndAdd(Long.SIZE);
    +      }
    +      if (timeBucket > bucketedValue.getTimeBucket()) {
    +
    +        if (bucketedValue.getValue() != null) {
    +          //overriding the value so need to subtract the size of previous value
    +          sizeInBytes.getAndAdd(-bucketedValue.getValue().length);
    +        }
    +
    +        sizeInBytes.getAndAdd(value.length);
    --- End diff --
    
    From @amberarrow 
    Can avoid 2 calls to getAndAdd() with something like:
    ```java
    int inc = null == bucketedValue.getValue() ? value.length : value.length - bucketedValue.getValue().length;
    sizeInBytes.getAndAdd(inc);
    ```


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1897 added managed ...

Posted by amberarrow <gi...@git.apache.org>.

Github user amberarrow commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r56767189
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/ManagedTimeStateImpl.java ---
    @@ -0,0 +1,125 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.util.concurrent.Future;
    +
    +import javax.validation.constraints.Min;
    +
    +import com.google.common.util.concurrent.Futures;
    +
    +import com.datatorrent.api.annotation.OperatorAnnotation;
    +import com.datatorrent.lib.state.BucketedState;
    +import com.datatorrent.lib.state.TimeSlicedBucketedState;
    +import com.datatorrent.netlet.util.Slice;
    +
    +/**
    + * This implementation of {@link ManagedState} lets the client to specify the time for each key. The value of time
    + * is used to derive the time-bucket of a key.
    + */
    +@OperatorAnnotation(checkpointableWithinAppWindow = false)
    +public class ManagedTimeStateImpl extends AbstractManagedStateImpl implements TimeSlicedBucketedState
    +{
    +  public ManagedTimeStateImpl()
    +  {
    +    this.numBuckets = 1;
    +  }
    +
    +  @Override
    +  public void put(long bucketId, long time, Slice key, Slice value)
    +  {
    +    int bucketIdx = prepareBucket(bucketId);
    +    long timeBucket = timeBucketAssigner.getTimeBucketFor(time);
    +    if (timeBucket != -1) {
    +      buckets[bucketIdx].put(key, timeBucket, value);
    +    }
    +  }
    +
    +  @Override
    +  public Slice getSync(long bucketId, Slice key)
    +  {
    +    int bucketIdx = prepareBucket(bucketId);
    +    return buckets[bucketIdx].get(key, -1, Bucket.ReadSource.ALL);
    +  }
    +
    +  @Override
    +  public Slice getSync(long bucketId, long time, Slice key)
    +  {
    +    int bucketIdx = prepareBucket(bucketId);
    +    long timeBucket = timeBucketAssigner.getTimeBucketFor(time);
    +    if (timeBucket == -1) {
    +      //time is expired so no point in looking further.
    +      return BucketedState.EXPIRED;
    +    }
    +    return buckets[bucketIdx].get(key, timeBucket, Bucket.ReadSource.ALL);
    +  }
    +
    +  @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
    +  @Override
    +  public Future<Slice> getAsync(long bucketId, Slice key)
    +  {
    +    int bucketIdx = prepareBucket(bucketId);
    +    Bucket bucket = buckets[bucketIdx];
    +    synchronized (bucket) {
    +      Slice cachedVal = buckets[bucketIdx].get(key, -1, Bucket.ReadSource.MEMORY);
    +      if (cachedVal != null) {
    +        return Futures.immediateFuture(cachedVal);
    +      }
    +      return readerService.submit(new KeyFetchTask(bucket, key, -1, throwable));
    +    }
    +  }
    +
    +  @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
    +  @Override
    +  public Future<Slice> getAsync(long bucketId, long time, Slice key)
    +  {
    +    int bucketIdx = prepareBucket(bucketId);
    +    Bucket bucket = buckets[bucketIdx];
    +    long timeBucket = timeBucketAssigner.getTimeBucketFor(time);
    +    if (timeBucket == -1) {
    +      //time is expired so no point in looking further.
    +      return Futures.immediateFuture(BucketedState.EXPIRED);
    +    }
    +    synchronized (bucket) {
    +      Slice cachedVal = buckets[bucketIdx].get(key, timeBucket, Bucket.ReadSource.MEMORY);
    +      if (cachedVal != null) {
    +        return Futures.immediateFuture(cachedVal);
    +      }
    +      return readerService.submit(new KeyFetchTask(bucket, key, timeBucket, throwable));
    +    }
    --- End diff --
    
    This synchronized block and the assignment to `bucket` are duplicated in multiple places (3 or 4); could it be factored out to a helper method in the base class ?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1897 added managed ...

Posted by amberarrow <gi...@git.apache.org>.

Github user amberarrow commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r56765586
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/ManagedTimeUnifiedStateImpl.java ---
    @@ -0,0 +1,238 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.io.DataInputStream;
    +import java.io.DataOutputStream;
    +import java.io.IOException;
    +import java.util.concurrent.Future;
    +import java.util.concurrent.LinkedBlockingQueue;
    +
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import org.apache.hadoop.fs.LocatedFileStatus;
    +import org.apache.hadoop.fs.RemoteIterator;
    +
    +import com.google.common.base.Preconditions;
    +import com.google.common.collect.Queues;
    +import com.google.common.util.concurrent.Futures;
    +
    +import com.datatorrent.lib.fileaccess.FileAccess;
    +import com.datatorrent.lib.state.BucketedState;
    +import com.datatorrent.netlet.util.Slice;
    +
    +/**
    + * In this implementation of {@link ManagedState} the buckets in memory are time-buckets.
    + * <p/>
    + *
    + * <b>Difference from {@link ManagedTimeStateImpl}</b>: <br/>
    + * <ol>
    + * <li>The main buckets in {@link ManagedTimeStateImpl} are unique adhoc long ids which the user provides with the
    + * key. In this implementation the main buckets are time buckets. The user provides just the time and the time bucket is
    + * derived from it.
    + * </li>
    + * <br/>
    + *
    + * <li>In regards to the bucket data on disk, in {@link ManagedTimeStateImpl} the buckets are persisted on disk
    + * with each bucket data further grouped into time-buckets: {base_path}/{bucketId}/{time-bucket id}. <br/>
    + * In this implementation operator id is used as bucketId (on disk) and there is just one time-bucket under a
    + * particular operator id:
    + * {base_path}/{operator id}/{time bucket id}.
    + * </li>
    + * <br/>
    + *
    + * <li>In {@link ManagedTimeStateImpl} a bucket belongs to just one partition. Multiple partitions cannot write to
    + * the same bucket. <br/>
    + * In this implementation multiple partitions can be working with the same time-bucket (since time-bucket is derived
    + * from time). This works because on the disk the time-bucket data is segregated under each operator id.
    + * </li>
    + * <br/>
    + *
    + * <li>While {@link ManagedTimeStateImpl} can support dynamic partitioning by pre-allocating buckets this will not
    + * be able to support dynamic partitioning efficiently.
    --- End diff --
    
    These contrasting statements should also include the third implementation class.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-apex-malhar pull request: MLHR-1897 #comment added manag...

Posted by chandnisingh <gi...@git.apache.org>.

Github user chandnisingh commented on a diff in the pull request:

    https://github.com/apache/incubator-apex-malhar/pull/145#discussion_r53717920
  
    --- Diff: library/src/main/java/com/datatorrent/lib/state/managed/TimeBucketAssigner.java ---
    @@ -0,0 +1,223 @@
    +/**
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you 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.datatorrent.lib.state.managed;
    +
    +import java.util.Calendar;
    +import java.util.Set;
    +
    +import javax.validation.constraints.NotNull;
    +
    +import org.joda.time.Duration;
    +
    +import com.esotericsoftware.kryo.serializers.FieldSerializer;
    +import com.esotericsoftware.kryo.serializers.JavaSerializer;
    +import com.google.common.collect.Sets;
    +
    +import com.datatorrent.api.Component;
    +import com.datatorrent.api.Context;
    +import com.datatorrent.lib.appdata.query.WindowBoundedService;
    +
    +/**
    + * Keeps track of time buckets.<br/>
    + *
    + * The data of a bucket is further divided into time-buckets. This component controls the length of time buckets,
    + * which time-bucket an event falls into and sliding the time boundaries.
    + * <p/>
    + *
    + * The configuration {@link #expireBefore} and {@link #bucketSpan}  are used to calculate number of time-buckets.
    + * For eg. if <code>expireBefore = 1 hour</code> and <code>bucketSpan = 30 minutes</code>, then <code>
    + *   numBuckets = 60 minutes/ 30 minutes = 2 </code>.
    + * <p/>
    + *
    + * The time boundaries- start and end, periodically move by span of a single time-bucket. Any event with time < start
    + * is expired. These boundaries slide between application window by another thread and not the operator thread.
    + */
    +public class TimeBucketAssigner implements Component<Context.OperatorContext>
    +{
    +  @NotNull
    +  @FieldSerializer.Bind(JavaSerializer.class)
    +  private Duration expireBefore = Duration.standardDays(2);
    +
    +  @FieldSerializer.Bind(JavaSerializer.class)
    +  private Duration bucketSpan;
    +
    +  private long bucketSpanMillis;
    +
    +  private long fixedStartTime;
    +  private long startTime;
    +  private long endTime;
    +  private int numBuckets;
    +
    +  private boolean initialized;
    +
    +  private transient WindowBoundedService windowBoundedService;
    +
    +  @NotNull
    +  private final transient Set<Listener> listeners = Sets.newHashSet();
    +
    +  private final transient Runnable expiryTask = new Runnable()
    +  {
    +    @Override
    +    public void run()
    +    {
    +      synchronized (lock) {
    +        startTime += bucketSpanMillis;
    +        endTime += bucketSpanMillis;
    +        for (Listener listener : listeners) {
    +          listener.purgeTimeBucketsBefore(startTime);
    +        }
    +      }
    +    }
    +  };
    +
    +  private final transient Object lock = new Object();
    +
    +  @Override
    +  public void setup(Context.OperatorContext context)
    +  {
    +    if (!initialized) {
    +      if (bucketSpan == null) {
    +        bucketSpan = Duration.millis(context.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT) *
    +            context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS));
    +      }
    +      Calendar calendar = Calendar.getInstance();
    +      long now = calendar.getTimeInMillis();
    +      fixedStartTime = now - expireBefore.getMillis();
    +      startTime = fixedStartTime;
    +
    +      bucketSpanMillis = bucketSpan.getMillis();
    +      numBuckets = (int)((expireBefore.getMillis() + bucketSpanMillis - 1) / bucketSpanMillis);
    +      endTime = startTime + (numBuckets * bucketSpanMillis);
    +
    +      initialized = true;
    +    }
    +    windowBoundedService = new WindowBoundedService(bucketSpanMillis, expiryTask);
    +    windowBoundedService.setup(context);
    +  }
    +
    +  public void beginWindow(long l)
    +  {
    +    windowBoundedService.beginWindow(l);
    +  }
    +
    +  public void endWindow()
    +  {
    +    windowBoundedService.endWindow();
    +  }
    +
    +  /**
    +   * Get the bucket key for the long value.
    +   *
    +   * @param value value from which bucket key is derived.
    +   * @return -1 if value is already expired; bucket key otherwise.
    +   */
    +  public long getTimeBucketFor(long value)
    +  {
    +    long lstart;
    +    long lend;
    +    synchronized (lock) {
    +      lstart = startTime;
    +      lend = endTime;
    +    }
    +    if (value < lstart) {
    +      return -1;
    +    }
    +    long diffFromStart = value - fixedStartTime;
    +    long key = diffFromStart / bucketSpanMillis;
    +    if (value > lend) {
    +      long move = ((value - lend) / bucketSpanMillis + 1) * bucketSpanMillis;
    +      synchronized (lock) {
    +        startTime = lstart + move;
    +        endTime = lend + move;
    +      }
    --- End diff --
    
    >> Can (value - lend) ever exceed bucketSpanMillis ?
    Yes. ```startTime``` is initialized during setup() and ```endTime = startTime + numBucket*bucketSpan```. 
    
    However ```value```  can come from the event. This time in the event can be ahead of ```endTime``` (we have seen this in a real application).  This triggers the whole boundary to move ahead.
    
    >> Second, it seems like this assignment could overwrite the assignment made to startTime and endTime by the expiryTask thread in some scenarios, for example: (a) We copy the values in the first synchronized block above (b) The expiryTask thread increments values by bucketSpanMillis (c) We overwrite those values by the values computed here. Is this OK ?
    
    We haven't seen a problem so far with this in BucketManager because the time in the event is given a priority over expiry task. However here there maybe an issue because the listener (```BucketsDataManager```) is notified to purge data before start time.
    
    Will club (a) and (c) in one block 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---