PhantomThief / simple-failover-java / 370

pending completion

Build # 370

Build Type

Pull #39

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/PhantomThief/simple-failover-java/commit/<a class=hub.com/PhantomThief/simple-failover-java/commit/<a class="double-link" href="https://git"><a class=hub.com/PhantomThief/simple-failover-java/commit/aa1c13bf9264e8e68857df97ce63008842d78cfd">aa1c13bf9">&lt;a href=&quot;https://github.com/PhantomThief/simple-failover-java/commit/</a><a class="double-link" href="https://github.com/PhantomThief/simple-failover-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/PhantomThief/simple-failover-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/PhantomThief/simple-failover-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/PhantomThief/simple-failover-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/PhantomThief/simple-failover-java/commit/aa1c13bf9264e8e68857df97ce63008842d78cfd">aa1c13bf9</a><a href="https://github.com/PhantomThief/simple-failover-java/commit/aa1c13bf9264e8e68857df97ce63008842d78cfd">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/PhantomThief/simple-failover-java/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/PhantomThief/simple-failover-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/PhantomThief/simple-failover-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/PhantomThief/simple-failover-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/PhantomThief/simple-failover-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/PhantomThief/simple-failover-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/PhantomThief/simple-failover-java/commit/aa1c13bf9264e8e68857df97ce63008... (continued)

Pull Request Pull Request #39: feat(SharedResource): 减小锁粒度

Run Details

90 of 90 new or added lines in 1 file covered. (100.0%)

1529 of 1823 relevant lines covered (83.87%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.8

/src/main/java/com/github/phantomthief/failover/util/SharedResource.java

package com.github.phantomthief.failover.util;

import static com.github.phantomthief.util.MoreSuppliers.lazy;
import static org.slf4j.LoggerFactory.getLogger;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

import org.slf4j.Logger;

import com.github.phantomthief.util.ThrowableConsumer;

/**
 * 多处使用的资源, 需要在所有使用者都注销之后, 再进行清理, 类似于引用计数.
 * <p>
 * TODO: 其实和netty的io.netty.util.ReferenceCounted很相似
 * 回头可以考虑使用和netty一样的方式，使用无锁的方式……
 * <p>
 * TODO: 回头挪到common-util包中
 *
 * @author w.vela
 * Created on 16/2/19.
 */
public class SharedResource<K, V> {

    private static final Logger logger = getLogger(SharedResource.class);

    private final ConcurrentMap<K, ResourceWrapper<K, V>> resourceMap = new ConcurrentHashMap<>();

    /**
     * 不管是否成功，都只get一次，然后缓存起来
     */
    private static class OnceSupplier<T> implements Supplier<T> {

        private enum Status {
            CREATED,
            ACTIVATED,
            BROKEN
        }

        private final Supplier<T> delegate;
        private final AtomicReference<OnceSupplier.Status> status = new AtomicReference<>(OnceSupplier.Status.CREATED);
        // 把 broken 的异常缓存起来，在后续的 get 时上抛
        private volatile Throwable brokenException;

        public OnceSupplier(Supplier<T> delegate) {
            this.delegate = lazy(delegate);
        }

        private synchronized T doInit() {
            if (status.get() == OnceSupplier.Status.ACTIVATED) {
                return delegate.get();
            } else if (status.get() == OnceSupplier.Status.BROKEN) {
                throw new RuntimeException("resource broken");
            }
            try {
                T resource = delegate.get();
                status.set(OnceSupplier.Status.ACTIVATED);
                return resource;
            } catch (Throwable t) {
                brokenException = t;
                status.set(OnceSupplier.Status.BROKEN);
                throw t;
            }
        }

        @Override
        public T get() {
            if (status.get() == OnceSupplier.Status.ACTIVATED) {
                return delegate.get();
            } else if (status.get() == OnceSupplier.Status.BROKEN) {
                throw new OnceBrokenException(brokenException);
            }
            return doInit();
        }
    }

    private static class ResourceWrapper<K, V> {

        private final K key;
        private final OnceSupplier<V> resourceSupplier;

        @javax.annotation.concurrent.GuardedBy("ResourceWrapper::this")
        private volatile int counter = 0;
        @GuardedBy("ResourceWrapper::this")
        private volatile boolean expired = false;

        public ResourceWrapper(K key, @Nonnull Supplier<V> resourceSupplier) {
            this.key = key;
            this.resourceSupplier = new OnceSupplier<>(resourceSupplier);
        }

        @Nonnull
        public V get() {
            return resourceSupplier.get();
        }

        public int count() {
            return counter;
        }

        public boolean incr() {
            synchronized (this) {
                if (!expired) {
                    counter++;
                    return true;
                }
            }
            return false;
        }

        public boolean decr() {
            synchronized (this) {
                if (counter <= 0) {
                    throw new AssertionError("INVALID INTERNAL STATE:" + key);
                }
                if (!expired) {
                    counter--;
                    if (counter <= 0) {
                        expired = true;
                    }
                    return true;
                }
            }
            return false;
        }
    }

    @Nonnull
    private ResourceWrapper<K, V> ensureWrapperExist(@Nonnull K key, @Nonnull Supplier<V> resource) {
        return resourceMap.compute(key, (k, v) -> {
            if (v == null || v.expired) {
                return new ResourceWrapper<>(k, resource);
            } else {
                return v;
            }
        });
    }

    @Nullable
    private ResourceWrapper<K, V> removeExpiredWrapper(@Nonnull K key) {
        AtomicReference<ResourceWrapper<K, V>> oldResourceWrapper = new AtomicReference<>(null);
        resourceMap.compute(key, (k, v) -> {
            if (v != null && v.expired) {
                oldResourceWrapper.set(v);
                return null;
            } else {
                return v;
            }
        });
        return oldResourceWrapper.get();
    }

    /**
     * 获取之前注册且未被注销的资源。
     * 使用相同 key 并发调用 get 和 register/unregister 时，本方法的返回值是不确定的（可能是null，也可能是之前注册的值）
     * 真实使用场景下，应避免相同的 key 并发 get/register/unregister，虽然本类是线程安全的，但是并发调用这些方法会让人很难理解
     * 真正共享的资源到底是哪一个
     */
    @Nullable
    public V get(@Nonnull K key) {
        ResourceWrapper<K, V> resourceWrapper = resourceMap.get(key);
        if (resourceWrapper != null) {
            return resourceWrapper.get();
        }
        return null;
    }

    /**
     * 这个方法实现的过程中，踩了挺多坑。这里把当前实现的关键思路记录一下，看后续有没有更好的实现方法:
     * 1. 如果之前成功创建过相同 key 的资源，就应当复用之前创建的资源；所以不能一上来就用 factory 创建资源
     * 2. 为了避免 unregister 已经在清理中的资源再次被增加引用计数，肯定需要加锁；为了避免全局锁，首先想到可以对 key 加锁
     * 3. 但是 key 有可能出现不同对象，但是 equals & hashcode 相同的情况，所以直接对 key 加锁会有问题
     * 4. key 不能用于加锁，那能不能对 value resourceWrapper 加锁呢？也有问题，因为为了避免资源泄漏，value 在引用计数减到 0 的时候
     * 需要删掉，这样导致有可能锁没有加在同一个对象上
     * 5. 所以目前采用这样的方式：我们在 resourceWrapper 上记录一个状态，如果某个资源的引用计数降到 0 则认为这个资源已经过期，
     * 不能继续使用/增加计数了；在 register 和 unregister 过程中发现已经过期的资源，采用清理 + 重建的方式来解决
     */
    @Nonnull
    public V register(@Nonnull K key, @Nonnull Function<K, V> factory) {
        boolean incr = false;
        ResourceWrapper<K, V> resourceWrapper = null;
        while (!incr) {
            resourceWrapper =
                    ensureWrapperExist(key, () -> Objects.requireNonNull(factory.apply(key), "factory 不应返回 null"));
            // 确保引用计数加成功才退出
            incr = resourceWrapper.incr();
        }
        try {
            // 尝试初始化资源
            return resourceWrapper.get();
        } catch (Throwable t) {
            // 初始化资源失败的话，把引用计数恢复一下，同时清理下 resourceMap 里引用计数降到 0 的记录
            unregister(key, v -> { });
            throw t;
        }
    }

    @Nonnull
    public V unregister(@Nonnull K key, @Nonnull ThrowableConsumer<V, Throwable> cleanup) {
        boolean decr = false;
        ResourceWrapper<K, V> resourceWrapper = null;
        while (!decr) {
            resourceWrapper = resourceMap.get(key);
            if (resourceWrapper == null) {
                throw new IllegalStateException("non paired unregister call for key:" + key);
            }
            decr = resourceWrapper.decr();
            removeExpiredWrapper(key);
        }
        V resource = resourceWrapper.get();
        if (resourceWrapper.expired) {
            try {
                cleanup.accept(resource);
                logger.info("cleanup resource:{}->{}", key, resource);
            } catch (Throwable e) {
                throw new UnregisterFailedException(e, resource);
            }
        }
        return resource;
    }

    public static class UnregisterFailedException extends RuntimeException {

        private final Object removed;

        private UnregisterFailedException(Throwable cause, Object removed) {
            super(cause);
            this.removed = removed;
        }

        @SuppressWarnings("unchecked")
        public <T> T getRemoved() {
            return (T) removed;
        }
    }

    public static class OnceBrokenException extends RuntimeException {

        private OnceBrokenException(Throwable cause) {
            super("resource broken", cause);
        }
    }
}

1	package com.github.phantomthief.failover.util;
2
3	import static com.github.phantomthief.util.MoreSuppliers.lazy;
4	import static org.slf4j.LoggerFactory.getLogger;
5
6	import java.util.Objects;
7	import java.util.concurrent.ConcurrentHashMap;
8	import java.util.concurrent.ConcurrentMap;
9	import java.util.concurrent.atomic.AtomicReference;
10	import java.util.function.Function;
11	import java.util.function.Supplier;
12
13	import javax.annotation.Nonnull;
14	import javax.annotation.Nullable;
15	import javax.annotation.concurrent.GuardedBy;
16
17	import org.slf4j.Logger;
18
19	import com.github.phantomthief.util.ThrowableConsumer;
20
21	/**
22	* 多处使用的资源, 需要在所有使用者都注销之后, 再进行清理, 类似于引用计数.
23	* <p>
24	* TODO: 其实和netty的io.netty.util.ReferenceCounted很相似
25	* 回头可以考虑使用和netty一样的方式，使用无锁的方式……
26	* <p>
27	* TODO: 回头挪到common-util包中
28	*
29	* @author w.vela
30	* Created on 16/2/19.
31	*/
32	public class SharedResource<K, V> {	1✔
33
34	private static final Logger logger = getLogger(SharedResource.class);	1✔
35
36	private final ConcurrentMap<K, ResourceWrapper<K, V>> resourceMap = new ConcurrentHashMap<>();	1✔
37
38	/**
39	* 不管是否成功，都只get一次，然后缓存起来
40	*/
41	private static class OnceSupplier<T> implements Supplier<T> {
42
43	private enum Status {	1✔
44	CREATED,	1✔
45	ACTIVATED,	1✔
46	BROKEN	1✔
47	}
48
49	private final Supplier<T> delegate;
50	private final AtomicReference<OnceSupplier.Status> status = new AtomicReference<>(OnceSupplier.Status.CREATED);	1✔
51	// 把 broken 的异常缓存起来，在后续的 get 时上抛
52	private volatile Throwable brokenException;
53
54	public OnceSupplier(Supplier<T> delegate) {	1✔
55	this.delegate = lazy(delegate);	1✔
56	}	1✔
57
58	private synchronized T doInit() {
59	if (status.get() == OnceSupplier.Status.ACTIVATED) {	1✔
60	return delegate.get();	×
61	} else if (status.get() == OnceSupplier.Status.BROKEN) {	1✔
62	throw new RuntimeException("resource broken");	×
63	}
64	try {
65	T resource = delegate.get();	1✔
66	status.set(OnceSupplier.Status.ACTIVATED);	1✔
67	return resource;	1✔
68	} catch (Throwable t) {	1✔
69	brokenException = t;	1✔
70	status.set(OnceSupplier.Status.BROKEN);	1✔
71	throw t;	1✔
72	}
73	}
74
75	@Override
76	public T get() {
77	if (status.get() == OnceSupplier.Status.ACTIVATED) {	1✔
78	return delegate.get();	1✔
79	} else if (status.get() == OnceSupplier.Status.BROKEN) {	1✔
80	throw new OnceBrokenException(brokenException);	1✔
81	}
82	return doInit();	1✔
83	}
84	}
85
86	private static class ResourceWrapper<K, V> {
87
88	private final K key;
89	private final OnceSupplier<V> resourceSupplier;
90
91	@javax.annotation.concurrent.GuardedBy("ResourceWrapper::this")	1✔
92	private volatile int counter = 0;
93	@GuardedBy("ResourceWrapper::this")	1✔
94	private volatile boolean expired = false;
95
96	public ResourceWrapper(K key, @Nonnull Supplier<V> resourceSupplier) {	1✔
97	this.key = key;	1✔
98	this.resourceSupplier = new OnceSupplier<>(resourceSupplier);	1✔
99	}	1✔
100
101	@Nonnull
102	public V get() {
103	return resourceSupplier.get();	1✔
104	}
105
106	public int count() {
107	return counter;	×
108	}
109
110	public boolean incr() {
111	synchronized (this) {	1✔
112	if (!expired) {	1✔
113	counter++;	1✔
114	return true;	1✔
115	}
116	}	×
117	return false;	×
118	}
119
120	public boolean decr() {
121	synchronized (this) {	1✔
122	if (counter <= 0) {	1✔
123	throw new AssertionError("INVALID INTERNAL STATE:" + key);	×
124	}
125	if (!expired) {	1✔
126	counter--;	1✔
127	if (counter <= 0) {	1✔
128	expired = true;	1✔
129	}
130	return true;	1✔
131	}
132	}	×
133	return false;	×
134	}
135	}
136
137	@Nonnull
138	private ResourceWrapper<K, V> ensureWrapperExist(@Nonnull K key, @Nonnull Supplier<V> resource) {
139	return resourceMap.compute(key, (k, v) -> {	1✔
140	if (v == null \|\| v.expired) {	1✔
141	return new ResourceWrapper<>(k, resource);	1✔
142	} else {
143	return v;	1✔
144	}
145	});
146	}
147
148	@Nullable
149	private ResourceWrapper<K, V> removeExpiredWrapper(@Nonnull K key) {
150	AtomicReference<ResourceWrapper<K, V>> oldResourceWrapper = new AtomicReference<>(null);	1✔
151	resourceMap.compute(key, (k, v) -> {	1✔
152	if (v != null && v.expired) {	1✔
153	oldResourceWrapper.set(v);	1✔
154	return null;	1✔
155	} else {
156	return v;	1✔
157	}
158	});
159	return oldResourceWrapper.get();	1✔
160	}
161
162	/**
163	* 获取之前注册且未被注销的资源。
164	* 使用相同 key 并发调用 get 和 register/unregister 时，本方法的返回值是不确定的（可能是null，也可能是之前注册的值）
165	* 真实使用场景下，应避免相同的 key 并发 get/register/unregister，虽然本类是线程安全的，但是并发调用这些方法会让人很难理解
166	* 真正共享的资源到底是哪一个
167	*/
168	@Nullable
169	public V get(@Nonnull K key) {
170	ResourceWrapper<K, V> resourceWrapper = resourceMap.get(key);	1✔
171	if (resourceWrapper != null) {	1✔
172	return resourceWrapper.get();	1✔
173	}
174	return null;	1✔
175	}
176
177	/**
178	* 这个方法实现的过程中，踩了挺多坑。这里把当前实现的关键思路记录一下，看后续有没有更好的实现方法:
179	* 1. 如果之前成功创建过相同 key 的资源，就应当复用之前创建的资源；所以不能一上来就用 factory 创建资源
180	* 2. 为了避免 unregister 已经在清理中的资源再次被增加引用计数，肯定需要加锁；为了避免全局锁，首先想到可以对 key 加锁
181	* 3. 但是 key 有可能出现不同对象，但是 equals & hashcode 相同的情况，所以直接对 key 加锁会有问题
182	* 4. key 不能用于加锁，那能不能对 value resourceWrapper 加锁呢？也有问题，因为为了避免资源泄漏，value 在引用计数减到 0 的时候
183	* 需要删掉，这样导致有可能锁没有加在同一个对象上
184	* 5. 所以目前采用这样的方式：我们在 resourceWrapper 上记录一个状态，如果某个资源的引用计数降到 0 则认为这个资源已经过期，
185	* 不能继续使用/增加计数了；在 register 和 unregister 过程中发现已经过期的资源，采用清理 + 重建的方式来解决
186	*/
187	@Nonnull
188	public V register(@Nonnull K key, @Nonnull Function<K, V> factory) {
189	boolean incr = false;	1✔
190	ResourceWrapper<K, V> resourceWrapper = null;	1✔
191	while (!incr) {	1✔
192	resourceWrapper =	1✔
193	ensureWrapperExist(key, () -> Objects.requireNonNull(factory.apply(key), "factory 不应返回 null"));	1✔
194	// 确保引用计数加成功才退出
195	incr = resourceWrapper.incr();	1✔
196	}
197	try {
198	// 尝试初始化资源
199	return resourceWrapper.get();	1✔
200	} catch (Throwable t) {	1✔
201	// 初始化资源失败的话，把引用计数恢复一下，同时清理下 resourceMap 里引用计数降到 0 的记录
202	unregister(key, v -> { });	×
203	throw t;	×
204	}
205	}
206
207	@Nonnull
208	public V unregister(@Nonnull K key, @Nonnull ThrowableConsumer<V, Throwable> cleanup) {
209	boolean decr = false;	1✔
210	ResourceWrapper<K, V> resourceWrapper = null;	1✔
211	while (!decr) {	1✔
212	resourceWrapper = resourceMap.get(key);	1✔
213	if (resourceWrapper == null) {	1✔
214	throw new IllegalStateException("non paired unregister call for key:" + key);	1✔
215	}
216	decr = resourceWrapper.decr();	1✔
217	removeExpiredWrapper(key);	1✔
218	}
219	V resource = resourceWrapper.get();	1✔
220	if (resourceWrapper.expired) {	1✔
221	try {
222	cleanup.accept(resource);	1✔
223	logger.info("cleanup resource:{}->{}", key, resource);	1✔
224	} catch (Throwable e) {	1✔
225	throw new UnregisterFailedException(e, resource);	1✔
226	}	1✔
227	}
228	return resource;	1✔
229	}
230
231	public static class UnregisterFailedException extends RuntimeException {
232
233	private final Object removed;
234
235	private UnregisterFailedException(Throwable cause, Object removed) {
236	super(cause);	1✔
237	this.removed = removed;	1✔
238	}	1✔
239
240	@SuppressWarnings("unchecked")
241	public <T> T getRemoved() {
242	return (T) removed;	1✔
243	}
244	}
245
246	public static class OnceBrokenException extends RuntimeException {
247
248	private OnceBrokenException(Throwable cause) {
249	super("resource broken", cause);	1✔
250	}	1✔
251	}
252	}

PhantomThief / simple-failover-java / 370

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous