Add Event and EventTest

remoting/android/java/src/org/chromium/chromoting/Event.java

+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package org.chromium.chromoting;
+
+import java.lang.ref.WeakReference;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+
+/**
+ * A thread-safe event queue which provides both {@link #add} and {@link #remove} functions with
+ * O(log(n)) time complex, and a {@link raise} function in Executor inheritance to execute all
+ * queued object.
+ *
+ * @param <ParamType> The parameter used in {@link ParameterRunnable} callback.
+ */
+public class Event<ParamType> {
+    /**
+     * A runnable with parameter.
+     */
+    public static interface ParameterRunnable<ParamType> {
+        void run(ParamType p);
+    }
+
+    /**
+     * A callback with parameter.
+     */
+    public static interface ParameterCallback<ReturnType, ParamType> {
+        ReturnType run(ParamType p);
+    }
+
+    /**
+     * An event provider version of {@link Event} implementation, provides {@link raise} function to
+     * execute appended {@link ParameterRunnable}.
+     */
+    public static final class Executor<ParamType> extends Event<ParamType> {
+        public void clear() {
+            synchronized (mLock) {
+                mArray.clear();
+                mFreeSet.clear();
+            }
+        }
+
+        /**
+         * Executes all queued {@link ParameterRunnable} with |parameter|, returns an integer of
+         * total callbacks executed. Note, if an 'add' function call is executing concurrently
+         * with the 'raise' function call, the newly added object may not be executed.
+         */
+        public int raise(ParamType parameter) {
+            int r = 0;
+            for (int i = 0; i < mArray.size(); i++) {

[Hzj_jie, 2016/05/23 23:39:14]

This is memory ordering safe, in Java, "After we exit a synchronized block, we
release the monitor, which has the effect of flushing the cache to main memory,
so that writes made by this thread can be visible to other threads." (Refer to
wikipedia, https://en.wikipedia.org/wiki/Java_concurrency)

[Lambros, 2016/05/24 17:56:41]

I think that quote is misleading, and you do need synchronization in this case.

Imagine that ArrayList.add() were implemented something like this:

void add(T foo) {
  int new_size = mSize + 1;

  mSize = some_random_CPU_register; // freeing up that register

  ... reallocate buffer, insert foo into it ...

  // Restore the correct size.
  mSize = new_size;
}

This would be a legal, conforming implementation. Nothing in the JMM prevents an
optimizing compiler (or JIT compiler, or CPU hardware) from executing this
equivalent code.

Calling ArrayList.size() at the wrong time, without synchronization, can then
return the content of some random CPU register or memory location.

See also: http://blogs.grammatech.com/fun-with-concurrency-problems

I suggest making a shallow copy of the array, with synchronization.
Then iterate over the copied array (no synchronization needed).

[Hzj_jie, 2016/05/24 18:24:38]

I totally agree we would get a random value without synchronized protection in
your scenario. But I doubt whether arraylist would be implemented like this. I
would implement it like this,

void add(T foo) {
  reallocateIfNeeded(mSize + 1);
  mArray[mSize++] = foo;
}

int size() {
  return mSize;
}

So if an add function is called concurrently with a size function call, the
worst case is, we get an out-of-date value. Indeed in multithreading
programming, the out-of-date definition is not so accurate. i.e. we cannot
confirm which thread reach the raced condition first.

So I have dug out a little bit, and found both implementation of openjdk
(http://goo.gl/BdKjk3) and android (https://goo.gl/8HnyIQ), both ensured mSize
won't be set to a random value during add operation.

+                ParameterRunnable<ParamType> e = mArray.get(i);
+                if (e != null) {
+                    e.run(parameter);
+                    r++;
+                }
+            }
+            return r;
+        }
+    }
+
+    /**
+     * A self removable {@link ParameterRunner}, uses a boolean {@link ParameterCallback} to decide
+     * whether removes self from {@link Event} or not.
+     */
+    private static class SelfRemovableParameterRunnable<ParamType>
+            implements ParameterRunnable<ParamType> {
+        private final ParameterCallback<Boolean, ParamType> mRef;
+        private final Event<ParamType> mOwner;
+        // This lock is used to make sure mId is correctly set before remove in run function. i.e.
+        // mOwner.add and assigment of mId need to be atomic.
+        private final Object mLock;
+        private final int mId;
+
+        private SelfRemovableParameterRunnable(Event<ParamType> owner,
+                                               ParameterCallback<Boolean, ParamType> callback) {
+            Preconditions.notNull(owner);
+            Preconditions.notNull(callback);
+            mRef = callback;
+            mOwner = owner;
+            mLock = new Object();
+            synchronized (mLock) {
+                mId = mOwner.add(this);
+            }
+            Preconditions.isTrue(mId >= 0);
+        }
+
+        public final void run(ParamType p) {
+            if (!mRef.run(p)) {
+                synchronized (mLock) {
+                    Preconditions.isTrue(mOwner.remove(mId));
+                }
+            }
+        }
+
+        // This class is self-contained, i.e. consumers do not need to have a reference of this
+        // instance, but all the logic is in new function.
+        public static final <ParamType> void create(
+                Event<ParamType> owner,
+                ParameterCallback<Boolean, ParamType> callback) {
+            new SelfRemovableParameterRunnable<ParamType>(owner, callback);
+        }
+    }
+
+    /**
+     * A weak reference based {@link ParameterRunner}, so a registered callback won't block GC to
+     * finalize the instance.
+     */
+    private static class WeakParameterRunnable<ParamType>
+            extends SelfRemovableParameterRunnable<ParamType> {
+        private WeakParameterRunnable(Event<ParamType> owner,
+                                      final WeakReference<ParameterRunnable<ParamType>> ref) {
+            super(owner, new ParameterCallback<Boolean, ParamType>() {
+                @Override
+                public Boolean run(ParamType param) {
+                    ParameterRunnable<ParamType> runnable = ref.get();
+                    if (runnable == null) {
+                        return false;
+                    }
+                    runnable.run(param);
+                    return true;
+                }
+            });
+        }
+
+        public static final <ParamType> void create(
+                Event<ParamType> owner,
+                ParameterRunnable<ParamType> runnable) {
+            new WeakParameterRunnable<ParamType>(owner, new WeakReference<>(runnable));
+        }
+    }
+
+    protected final ArrayList<ParameterRunnable<ParamType>> mArray;
+    protected final HashSet<Integer> mFreeSet;
+    protected final Object mLock;
+
+    public Event() {
+        mArray = new ArrayList<>();
+        mFreeSet = new HashSet<>();
+        mLock = new Object();
+    }
+
+    /**
+     * Adds a {@link ParameterRunnable} object into current instance, returns an integer as its
+     * position.
+     */
+    public int add(ParameterRunnable<ParamType> runnable) {
+        if (runnable == null) {
+            return -1;
+        }
+        synchronized (mLock) {
+            if (mFreeSet.isEmpty()) {
+                Preconditions.isTrue(mArray.add(runnable));
+                return mArray.size() - 1;
+            }
+            Iterator<Integer> it = mFreeSet.iterator();
+            Preconditions.isTrue(it.hasNext());
+            int id = it.next();
+            mArray.set(id, runnable);
+            Preconditions.isTrue(mFreeSet.remove(id));
+            return id;
+        }
+    }
+
+    /**
+     * Adds a self removable {@link ParameterRunnable} object into current instance, since
+     * {@link SelfRemovableParameterRunnable} will remove itself from {@link Event} instance, this
+     * function returns a boolean value to indicate the success or not of the operation.
+     */
+    public boolean addSelfRemovable(ParameterCallback<Boolean, ParamType> callback) {
+        if (callback == null) {
+            return false;
+        }
+        SelfRemovableParameterRunnable.<ParamType>create(this, callback);
+        return true;
+    }
+
+    /**
+     * Adds a weak referred {@link ParameterRunnable} object into current instance, since
+     * {@link WeakParameterRunnable} will remove itself from {@link Event} instance, this function
+     * returns a boolean value to indicate the success or not of the operation.
+     */
+    public boolean addWeakReferred(ParameterRunnable<ParamType> runnable) {
+        if (runnable == null) {
+            return false;
+        }
+        WeakParameterRunnable.<ParamType>create(this, runnable);
+        return true;
+    }
+
+    /**
+     * Removes a Runnable object at position |id|, returns false if the |id| is not in the range of
+     * {@link mArray}.
+     */
+    public boolean remove(int id) {
+        if (id < 0 || id >= mArray.size() || mArray.get(id) == null) {
+            return false;
+        }
+        synchronized (mLock) {
+            mArray.set(id, null);
+            mFreeSet.add(id);
+        }
+        return true;
+    }
+
+    /**
+     * Returns the total count of runnables attached to current instance.
+     */
+    public int size() {
+        synchronized (mLock) {
+            return mArray.size() - mFreeSet.size();
+        }
+    }
+
+    /**
+     * Returns true if there are at least one runnable attached to current instance.
+     */
+    public boolean attached() {
+        return size() > 0;
+    }
+}