Committed by
Madan Jampani
Updating multimap API and commands and providing implementation.
Change-Id: Iff49b429cfc7c0142f3ab2e1dde1a32e85f20e87 (cherry picked from commit 44a1fef9)
Showing
7 changed files
with
1615 additions
and
96 deletions
... | @@ -16,7 +16,6 @@ | ... | @@ -16,7 +16,6 @@ |
16 | 16 | ||
17 | package org.onosproject.store.service; | 17 | package org.onosproject.store.service; |
18 | 18 | ||
19 | -import com.google.common.collect.Multimap; | ||
20 | import com.google.common.collect.Multiset; | 19 | import com.google.common.collect.Multiset; |
21 | 20 | ||
22 | import java.util.Collection; | 21 | import java.util.Collection; |
... | @@ -92,7 +91,7 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { | ... | @@ -92,7 +91,7 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { |
92 | * and others ignoring put requests for existing entries. | 91 | * and others ignoring put requests for existing entries. |
93 | * @param key the key to add | 92 | * @param key the key to add |
94 | * @param value the value to add | 93 | * @param value the value to add |
95 | - * @return a future whose value will betrue if the map has changed because | 94 | + * @return a future whose value will be true if the map has changed because |
96 | * of this call, false otherwise | 95 | * of this call, false otherwise |
97 | */ | 96 | */ |
98 | CompletableFuture<Boolean> put(K key, V value); | 97 | CompletableFuture<Boolean> put(K key, V value); |
... | @@ -119,16 +118,18 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { | ... | @@ -119,16 +118,18 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { |
119 | * @return a future whose value will be true if the map changes because of | 118 | * @return a future whose value will be true if the map changes because of |
120 | * this call, false otherwise. | 119 | * this call, false otherwise. |
121 | */ | 120 | */ |
122 | - CompletableFuture<Boolean> removeAll(K key, Iterable<? extends V> values); | 121 | + CompletableFuture<Boolean> removeAll(K key, |
122 | + Collection<? extends V> values); | ||
123 | 123 | ||
124 | /** | 124 | /** |
125 | * Removes all values associated with the specified key as well as the key | 125 | * Removes all values associated with the specified key as well as the key |
126 | * itself. | 126 | * itself. |
127 | * @param key the key whose key-value pairs will be removed | 127 | * @param key the key whose key-value pairs will be removed |
128 | * @return a future whose value is the set of values that were removed, | 128 | * @return a future whose value is the set of values that were removed, |
129 | - * which may be empty | 129 | + * which may be empty, if the values did not exist the version will be |
130 | + * less than one. | ||
130 | */ | 131 | */ |
131 | - CompletableFuture<Versioned<Collection<byte[]>>> removeAll(K key); | 132 | + CompletableFuture<Versioned<Collection<? extends V>>> removeAll(K key); |
132 | 133 | ||
133 | /** | 134 | /** |
134 | * Adds the set of key-value pairs of the specified key with each of the | 135 | * Adds the set of key-value pairs of the specified key with each of the |
... | @@ -140,17 +141,8 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { | ... | @@ -140,17 +141,8 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { |
140 | * @return a future whose value will be true if any change in the map | 141 | * @return a future whose value will be true if any change in the map |
141 | * results from this call, false otherwise | 142 | * results from this call, false otherwise |
142 | */ | 143 | */ |
143 | - CompletableFuture<Boolean> putAll(K key, Iterable<? extends V> values); | 144 | + CompletableFuture<Boolean> putAll(K key, |
144 | - | 145 | + Collection<? extends V> values); |
145 | - /** | ||
146 | - * Adds all entries from this multimap that are not already present, and | ||
147 | - * may or may not add duplicate entries depending on the implementation. | ||
148 | - * @param multiMap the map whose entries should be added | ||
149 | - * @return a future whose value will be true if any change results from | ||
150 | - * this call, false otherwise | ||
151 | - */ | ||
152 | - CompletableFuture<Boolean> putAll( | ||
153 | - Multimap<? extends K, ? extends V> multiMap); | ||
154 | 146 | ||
155 | /** | 147 | /** |
156 | * Stores all the values in values associated with the key specified, | 148 | * Stores all the values in values associated with the key specified, |
... | @@ -161,7 +153,8 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { | ... | @@ -161,7 +153,8 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { |
161 | * @return a future whose value will be the collection of removed values, | 153 | * @return a future whose value will be the collection of removed values, |
162 | * which may be empty | 154 | * which may be empty |
163 | */ | 155 | */ |
164 | - CompletableFuture<Collection<V>> replaceValues(K key, Iterable<V> values); | 156 | + CompletableFuture<Versioned<Collection<? extends V>>> replaceValues( |
157 | + K key, Collection<V> values); | ||
165 | 158 | ||
166 | /** | 159 | /** |
167 | * Removes all key-value pairs, after which it will be empty. | 160 | * Removes all key-value pairs, after which it will be empty. |
... | @@ -177,7 +170,7 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { | ... | @@ -177,7 +170,7 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { |
177 | * @return a future whose value will be the collection of the values | 170 | * @return a future whose value will be the collection of the values |
178 | * associated with the specified key, the collection may be empty | 171 | * associated with the specified key, the collection may be empty |
179 | */ | 172 | */ |
180 | - CompletableFuture<Collection<V>> get(K key); | 173 | + CompletableFuture<Versioned<Collection<? extends V>>> get(K key); |
181 | 174 | ||
182 | /** | 175 | /** |
183 | * Returns a set of the keys contained in this multimap with one or more | 176 | * Returns a set of the keys contained in this multimap with one or more |
... | @@ -203,7 +196,7 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { | ... | @@ -203,7 +196,7 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { |
203 | * @return a future whose value will be a collection of values, this may be | 196 | * @return a future whose value will be a collection of values, this may be |
204 | * empty | 197 | * empty |
205 | */ | 198 | */ |
206 | - CompletableFuture<Collection<V>> values(); | 199 | + CompletableFuture<Multiset<V>> values(); |
207 | 200 | ||
208 | /** | 201 | /** |
209 | * Returns a collection of each key-value pair in this map. | 202 | * Returns a collection of each key-value pair in this map. | ... | ... |
... | @@ -25,6 +25,7 @@ COMPILE_DEPS = [ | ... | @@ -25,6 +25,7 @@ COMPILE_DEPS = [ |
25 | TEST_DEPS = [ | 25 | TEST_DEPS = [ |
26 | '//lib:TEST', | 26 | '//lib:TEST', |
27 | '//core/api:onos-api-tests', | 27 | '//core/api:onos-api-tests', |
28 | + '//lib:onos-atomix', | ||
28 | ] | 29 | ] |
29 | 30 | ||
30 | osgi_jar_with_tests ( | 31 | osgi_jar_with_tests ( | ... | ... |
... | @@ -28,9 +28,9 @@ import io.atomix.catalyst.util.Assert; | ... | @@ -28,9 +28,9 @@ import io.atomix.catalyst.util.Assert; |
28 | import io.atomix.copycat.Command; | 28 | import io.atomix.copycat.Command; |
29 | import io.atomix.copycat.Query; | 29 | import io.atomix.copycat.Query; |
30 | import org.onlab.util.Match; | 30 | import org.onlab.util.Match; |
31 | +import org.onosproject.store.service.Versioned; | ||
31 | 32 | ||
32 | import java.util.Collection; | 33 | import java.util.Collection; |
33 | -import java.util.List; | ||
34 | import java.util.Map; | 34 | import java.util.Map; |
35 | import java.util.Set; | 35 | import java.util.Set; |
36 | 36 | ||
... | @@ -123,7 +123,8 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -123,7 +123,8 @@ public final class AsyncConsistentMultimapCommands { |
123 | } | 123 | } |
124 | 124 | ||
125 | @Override | 125 | @Override |
126 | - public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | 126 | + public void writeObject(BufferOutput<?> buffer, |
127 | + Serializer serializer) { | ||
127 | super.writeObject(buffer, serializer); | 128 | super.writeObject(buffer, serializer); |
128 | serializer.writeObject(key, buffer); | 129 | serializer.writeObject(key, buffer); |
129 | } | 130 | } |
... | @@ -166,7 +167,8 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -166,7 +167,8 @@ public final class AsyncConsistentMultimapCommands { |
166 | } | 167 | } |
167 | 168 | ||
168 | @Override | 169 | @Override |
169 | - public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | 170 | + public void writeObject(BufferOutput<?> buffer, |
171 | + Serializer serializer) { | ||
170 | super.writeObject(buffer, serializer); | 172 | super.writeObject(buffer, serializer); |
171 | } | 173 | } |
172 | 174 | ||
... | @@ -265,49 +267,221 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -265,49 +267,221 @@ public final class AsyncConsistentMultimapCommands { |
265 | } | 267 | } |
266 | 268 | ||
267 | /** | 269 | /** |
268 | - * Update and get command. Note that corresponding values must have the | 270 | + * Remove command, backs remove and removeAll's that return booleans. |
269 | - * same index in the respective arrays. | ||
270 | */ | 271 | */ |
271 | @SuppressWarnings("serial") | 272 | @SuppressWarnings("serial") |
272 | - public static class UpdateAndGet extends | 273 | + public static class RemoveAll extends |
273 | - MultimapCommand<MapEntryUpdateResult<String, Collection<byte[]>>> { | 274 | + MultimapCommand<Versioned<Collection<? extends byte[]>>> { |
274 | private String key; | 275 | private String key; |
275 | - private List<byte[]> values; | 276 | + private Match<Long> versionMatch; |
276 | - private List<Match<byte[]>> valueMatches; | ||
277 | - private List<Match<Long>> versionMatches; | ||
278 | 277 | ||
279 | - public UpdateAndGet() { | 278 | + public RemoveAll() { |
280 | } | 279 | } |
281 | 280 | ||
282 | - public UpdateAndGet(String key, List<byte[]> values, | 281 | + public RemoveAll(String key, Match<Long> versionMatch) { |
283 | - List<Match<byte[]>> valueMatches, | 282 | + this.key = Assert.notNull(key, "key"); |
284 | - List<Match<Long>> versionMatches) { | 283 | + this.versionMatch = versionMatch; |
285 | - this.key = key; | 284 | + } |
286 | - this.values = values; | 285 | + |
287 | - this.valueMatches = valueMatches; | 286 | + public String key() { |
288 | - this.versionMatches = versionMatches; | 287 | + return this.key; |
288 | + } | ||
289 | + | ||
290 | + public Match<Long> versionMatch() { | ||
291 | + return versionMatch; | ||
292 | + } | ||
293 | + | ||
294 | + @Override | ||
295 | + public CompactionMode compaction() { | ||
296 | + return CompactionMode.FULL; | ||
297 | + } | ||
298 | + | ||
299 | + @Override | ||
300 | + public void writeObject(BufferOutput<?> buffer, | ||
301 | + Serializer serializer) { | ||
302 | + super.writeObject(buffer, serializer); | ||
303 | + serializer.writeObject(key, buffer); | ||
304 | + serializer.writeObject(versionMatch, buffer); | ||
305 | + } | ||
306 | + | ||
307 | + @Override | ||
308 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
309 | + super.readObject(buffer, serializer); | ||
310 | + key = serializer.readObject(buffer); | ||
311 | + versionMatch = serializer.readObject(buffer); | ||
312 | + } | ||
313 | + | ||
314 | + @Override | ||
315 | + public String toString() { | ||
316 | + return MoreObjects.toStringHelper(getClass()) | ||
317 | + .add("key", key) | ||
318 | + .add("versionMatch", versionMatch) | ||
319 | + .toString(); | ||
320 | + } | ||
321 | + } | ||
322 | + | ||
323 | + /** | ||
324 | + * Remove command, backs remove and removeAll's that return booleans. | ||
325 | + */ | ||
326 | + @SuppressWarnings("serial") | ||
327 | + public static class MultiRemove extends | ||
328 | + MultimapCommand<Boolean> { | ||
329 | + private String key; | ||
330 | + private Collection<byte[]> values; | ||
331 | + private Match<Long> versionMatch; | ||
332 | + | ||
333 | + public MultiRemove() { | ||
334 | + } | ||
335 | + | ||
336 | + public MultiRemove(String key, Collection<byte[]> valueMatches, | ||
337 | + Match<Long> versionMatch) { | ||
338 | + this.key = Assert.notNull(key, "key"); | ||
339 | + this.values = valueMatches; | ||
340 | + this.versionMatch = versionMatch; | ||
289 | } | 341 | } |
290 | 342 | ||
291 | public String key() { | 343 | public String key() { |
292 | return this.key; | 344 | return this.key; |
293 | } | 345 | } |
294 | 346 | ||
295 | - public List<byte[]> values() { | 347 | + public Collection<byte[]> values() { |
348 | + return values; | ||
349 | + } | ||
350 | + | ||
351 | + public Match<Long> versionMatch() { | ||
352 | + return versionMatch; | ||
353 | + } | ||
354 | + | ||
355 | + @Override | ||
356 | + public CompactionMode compaction() { | ||
357 | + return CompactionMode.FULL; | ||
358 | + } | ||
359 | + | ||
360 | + @Override | ||
361 | + public void writeObject(BufferOutput<?> buffer, | ||
362 | + Serializer serializer) { | ||
363 | + super.writeObject(buffer, serializer); | ||
364 | + serializer.writeObject(key, buffer); | ||
365 | + serializer.writeObject(values, buffer); | ||
366 | + serializer.writeObject(versionMatch, buffer); | ||
367 | + } | ||
368 | + | ||
369 | + @Override | ||
370 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
371 | + super.readObject(buffer, serializer); | ||
372 | + key = serializer.readObject(buffer); | ||
373 | + values = serializer.readObject(buffer); | ||
374 | + versionMatch = serializer.readObject(buffer); | ||
375 | + } | ||
376 | + | ||
377 | + @Override | ||
378 | + public String toString() { | ||
379 | + return MoreObjects.toStringHelper(getClass()) | ||
380 | + .add("key", key) | ||
381 | + .add("values", values) | ||
382 | + .add("versionMatch", versionMatch) | ||
383 | + .toString(); | ||
384 | + } | ||
385 | + } | ||
386 | + | ||
387 | + /** | ||
388 | + * Command to back the put and putAll methods. | ||
389 | + */ | ||
390 | + @SuppressWarnings("serial") | ||
391 | + public static class Put extends MultimapCommand<Boolean> { | ||
392 | + private String key; | ||
393 | + private Collection<? extends byte[]> values; | ||
394 | + private Match<Long> versionMatch; | ||
395 | + | ||
396 | + public Put() { | ||
397 | + } | ||
398 | + | ||
399 | + public Put(String key, Collection<? extends byte[]> values, | ||
400 | + Match<Long> versionMatch) { | ||
401 | + this.key = Assert.notNull(key, "key"); | ||
402 | + this.values = values; | ||
403 | + this.versionMatch = versionMatch; | ||
404 | + } | ||
405 | + | ||
406 | + public String key() { | ||
407 | + return key; | ||
408 | + } | ||
409 | + | ||
410 | + public Collection<? extends byte[]> values() { | ||
296 | return values; | 411 | return values; |
297 | } | 412 | } |
298 | 413 | ||
299 | - public List<Match<byte[]>> valueMatches() { | 414 | + public Match<Long> versionMatch() { |
300 | - return valueMatches; | 415 | + return versionMatch; |
416 | + } | ||
417 | + | ||
418 | + @Override | ||
419 | + public CompactionMode compaction() { | ||
420 | + return CompactionMode.QUORUM; | ||
421 | + } | ||
422 | + | ||
423 | + @Override | ||
424 | + public void writeObject(BufferOutput<?> buffer, | ||
425 | + Serializer serializer) { | ||
426 | + super.writeObject(buffer, serializer); | ||
427 | + serializer.writeObject(key, buffer); | ||
428 | + serializer.writeObject(values, buffer); | ||
429 | + serializer.writeObject(versionMatch, buffer); | ||
430 | + } | ||
431 | + | ||
432 | + @Override | ||
433 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
434 | + super.readObject(buffer, serializer); | ||
435 | + key = serializer.readObject(buffer); | ||
436 | + values = serializer.readObject(buffer); | ||
437 | + versionMatch = serializer.readObject(buffer); | ||
438 | + } | ||
439 | + | ||
440 | + @Override | ||
441 | + public String toString() { | ||
442 | + return MoreObjects.toStringHelper(getClass()) | ||
443 | + .add("key", key) | ||
444 | + .add("values", values) | ||
445 | + .add("versionMatch", versionMatch) | ||
446 | + .toString(); | ||
447 | + } | ||
448 | + } | ||
449 | + | ||
450 | + /** | ||
451 | + * Replace command, returns the collection that was replaced. | ||
452 | + */ | ||
453 | + @SuppressWarnings("serial") | ||
454 | + public static class Replace extends | ||
455 | + MultimapCommand<Versioned<Collection<? extends byte[]>>> { | ||
456 | + private String key; | ||
457 | + private Collection<byte[]> values; | ||
458 | + private Match<Long> versionMatch; | ||
459 | + | ||
460 | + public Replace() { | ||
461 | + } | ||
462 | + | ||
463 | + public Replace(String key, Collection<byte[]> values, | ||
464 | + Match<Long> versionMatch) { | ||
465 | + this.key = Assert.notNull(key, "key"); | ||
466 | + this.values = values; | ||
467 | + this.versionMatch = versionMatch; | ||
468 | + } | ||
469 | + | ||
470 | + public String key() { | ||
471 | + return this.key; | ||
301 | } | 472 | } |
302 | 473 | ||
303 | - public List<Match<Long>> versionMatches() { | 474 | + public Match<Long> versionMatch() { |
304 | - return versionMatches; | 475 | + return versionMatch; |
476 | + } | ||
477 | + | ||
478 | + public Collection<byte[]> values() { | ||
479 | + return values; | ||
305 | } | 480 | } |
306 | 481 | ||
307 | @Override | 482 | @Override |
308 | public CompactionMode compaction() { | 483 | public CompactionMode compaction() { |
309 | - return values == null ? CompactionMode.FULL : | 484 | + return CompactionMode.FULL; |
310 | - CompactionMode.QUORUM; | ||
311 | } | 485 | } |
312 | 486 | ||
313 | @Override | 487 | @Override |
... | @@ -316,8 +490,7 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -316,8 +490,7 @@ public final class AsyncConsistentMultimapCommands { |
316 | super.writeObject(buffer, serializer); | 490 | super.writeObject(buffer, serializer); |
317 | serializer.writeObject(key, buffer); | 491 | serializer.writeObject(key, buffer); |
318 | serializer.writeObject(values, buffer); | 492 | serializer.writeObject(values, buffer); |
319 | - serializer.writeObject(valueMatches, buffer); | 493 | + serializer.writeObject(versionMatch, buffer); |
320 | - serializer.writeObject(versionMatches, buffer); | ||
321 | } | 494 | } |
322 | 495 | ||
323 | @Override | 496 | @Override |
... | @@ -325,13 +498,16 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -325,13 +498,16 @@ public final class AsyncConsistentMultimapCommands { |
325 | super.readObject(buffer, serializer); | 498 | super.readObject(buffer, serializer); |
326 | key = serializer.readObject(buffer); | 499 | key = serializer.readObject(buffer); |
327 | values = serializer.readObject(buffer); | 500 | values = serializer.readObject(buffer); |
328 | - valueMatches = serializer.readObject(buffer); | 501 | + versionMatch = serializer.readObject(buffer); |
329 | - versionMatches = serializer.readObject(buffer); | ||
330 | } | 502 | } |
331 | 503 | ||
332 | @Override | 504 | @Override |
333 | public String toString() { | 505 | public String toString() { |
334 | - return super.toString(); | 506 | + return MoreObjects.toStringHelper(getClass()) |
507 | + .add("key", key) | ||
508 | + .add("values", values) | ||
509 | + .add("versionMatch", versionMatch) | ||
510 | + .toString(); | ||
335 | } | 511 | } |
336 | } | 512 | } |
337 | 513 | ||
... | @@ -360,7 +536,7 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -360,7 +536,7 @@ public final class AsyncConsistentMultimapCommands { |
360 | * Value collection query. | 536 | * Value collection query. |
361 | */ | 537 | */ |
362 | @SuppressWarnings("serial") | 538 | @SuppressWarnings("serial") |
363 | - public static class Values extends MultimapQuery<Collection<byte[]>> { | 539 | + public static class Values extends MultimapQuery<Multiset<byte[]>> { |
364 | } | 540 | } |
365 | 541 | ||
366 | /** | 542 | /** |
... | @@ -374,7 +550,11 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -374,7 +550,11 @@ public final class AsyncConsistentMultimapCommands { |
374 | /** | 550 | /** |
375 | * Get value query. | 551 | * Get value query. |
376 | */ | 552 | */ |
377 | - public static class Get extends KeyQuery<Collection<byte[]>> { | 553 | + public static class Get extends |
554 | + KeyQuery<Versioned<Collection<? extends byte[]>>> { | ||
555 | + public Get(String key) { | ||
556 | + super(key); | ||
557 | + } | ||
378 | } | 558 | } |
379 | 559 | ||
380 | /** | 560 | /** |
... | @@ -387,7 +567,7 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -387,7 +567,7 @@ public final class AsyncConsistentMultimapCommands { |
387 | registry.register(ContainsKey.class, -1000); | 567 | registry.register(ContainsKey.class, -1000); |
388 | registry.register(ContainsValue.class, -1001); | 568 | registry.register(ContainsValue.class, -1001); |
389 | registry.register(ContainsEntry.class, -1002); | 569 | registry.register(ContainsEntry.class, -1002); |
390 | - registry.register(UpdateAndGet.class, -1003); | 570 | + registry.register(Replace.class, -1003); |
391 | registry.register(Clear.class, -1004); | 571 | registry.register(Clear.class, -1004); |
392 | registry.register(KeySet.class, -1005); | 572 | registry.register(KeySet.class, -1005); |
393 | registry.register(Keys.class, -1006); | 573 | registry.register(Keys.class, -1006); |
... | @@ -396,6 +576,9 @@ public final class AsyncConsistentMultimapCommands { | ... | @@ -396,6 +576,9 @@ public final class AsyncConsistentMultimapCommands { |
396 | registry.register(Size.class, -1009); | 576 | registry.register(Size.class, -1009); |
397 | registry.register(IsEmpty.class, -1010); | 577 | registry.register(IsEmpty.class, -1010); |
398 | registry.register(Get.class, -1011); | 578 | registry.register(Get.class, -1011); |
579 | + registry.register(Put.class, -1012); | ||
580 | + registry.register(RemoveAll.class, -1013); | ||
581 | + registry.register(MultiRemove.class, -1014); | ||
399 | } | 582 | } |
400 | } | 583 | } |
401 | } | 584 | } | ... | ... |
... | @@ -17,11 +17,10 @@ | ... | @@ -17,11 +17,10 @@ |
17 | package org.onosproject.store.primitives.resources.impl; | 17 | package org.onosproject.store.primitives.resources.impl; |
18 | 18 | ||
19 | import com.google.common.collect.Lists; | 19 | import com.google.common.collect.Lists; |
20 | -import com.google.common.collect.Multimap; | ||
21 | import com.google.common.collect.Multiset; | 20 | import com.google.common.collect.Multiset; |
22 | import io.atomix.copycat.client.CopycatClient; | 21 | import io.atomix.copycat.client.CopycatClient; |
23 | import io.atomix.resource.AbstractResource; | 22 | import io.atomix.resource.AbstractResource; |
24 | -import org.onlab.util.Match; | 23 | +import io.atomix.resource.ResourceTypeInfo; |
25 | import org.onosproject.store.service.AsyncConsistentMultimap; | 24 | import org.onosproject.store.service.AsyncConsistentMultimap; |
26 | import org.onosproject.store.service.Versioned; | 25 | import org.onosproject.store.service.Versioned; |
27 | 26 | ||
... | @@ -32,13 +31,28 @@ import java.util.Properties; | ... | @@ -32,13 +31,28 @@ import java.util.Properties; |
32 | import java.util.Set; | 31 | import java.util.Set; |
33 | import java.util.concurrent.CompletableFuture; | 32 | import java.util.concurrent.CompletableFuture; |
34 | 33 | ||
35 | -import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.*; | 34 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Clear; |
35 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.ContainsEntry; | ||
36 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.ContainsKey; | ||
37 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.ContainsValue; | ||
38 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Entries; | ||
39 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Get; | ||
40 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.IsEmpty; | ||
41 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.KeySet; | ||
42 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Keys; | ||
43 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.MultiRemove; | ||
44 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Put; | ||
45 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.RemoveAll; | ||
46 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Replace; | ||
47 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Size; | ||
48 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Values; | ||
36 | 49 | ||
37 | /** | 50 | /** |
38 | * Set based implementation of the {@link AsyncConsistentMultimap}. | 51 | * Set based implementation of the {@link AsyncConsistentMultimap}. |
39 | * <p> | 52 | * <p> |
40 | * Note: this implementation does not allow null entries or duplicate entries. | 53 | * Note: this implementation does not allow null entries or duplicate entries. |
41 | */ | 54 | */ |
55 | +@ResourceTypeInfo(id = -153, factory = AsyncConsistentSetMultimapFactory.class) | ||
42 | public class AsyncConsistentSetMultimap | 56 | public class AsyncConsistentSetMultimap |
43 | extends AbstractResource<AsyncConsistentSetMultimap> | 57 | extends AbstractResource<AsyncConsistentSetMultimap> |
44 | implements AsyncConsistentMultimap<String, byte[]> { | 58 | implements AsyncConsistentMultimap<String, byte[]> { |
... | @@ -81,68 +95,50 @@ public class AsyncConsistentSetMultimap | ... | @@ -81,68 +95,50 @@ public class AsyncConsistentSetMultimap |
81 | 95 | ||
82 | @Override | 96 | @Override |
83 | public CompletableFuture<Boolean> put(String key, byte[] value) { | 97 | public CompletableFuture<Boolean> put(String key, byte[] value) { |
84 | - return submit(new UpdateAndGet(key, Lists.newArrayList(value), | 98 | + return submit(new Put(key, Lists.newArrayList(value), null)); |
85 | - Lists.newArrayList(Match.NULL), | ||
86 | - Lists.newArrayList(Match.NULL))) | ||
87 | - .whenComplete((result, e) -> throwIfLocked(result.status())) | ||
88 | - .thenApply(result -> | ||
89 | - result.status() == MapEntryUpdateResult.Status.OK); | ||
90 | } | 99 | } |
91 | 100 | ||
92 | @Override | 101 | @Override |
93 | public CompletableFuture<Boolean> remove(String key, byte[] value) { | 102 | public CompletableFuture<Boolean> remove(String key, byte[] value) { |
94 | - return submit(new UpdateAndGet(key, Lists.newArrayList(value), | 103 | + return submit(new MultiRemove(key, |
95 | - Lists.newArrayList(Match.ifValue(value)), | 104 | + Lists.newArrayList(value), |
96 | - Lists.newArrayList(Match.NULL))) | 105 | + null)); |
97 | - .whenComplete((result, e) -> throwIfLocked(result.status())) | ||
98 | - .thenApply(result -> | ||
99 | - result.status() == MapEntryUpdateResult.Status.OK); | ||
100 | } | 106 | } |
101 | 107 | ||
102 | @Override | 108 | @Override |
103 | - public CompletableFuture<Boolean> removeAll(String key, Iterable<? extends byte[]> values) { | 109 | + public CompletableFuture<Boolean> removeAll( |
104 | - | 110 | + String key, Collection<? extends byte[]> values) { |
105 | - throw new UnsupportedOperationException("This operation cannot be " + | 111 | + return submit(new MultiRemove(key, (Collection<byte[]>) values, null)); |
106 | - "used without support for " + | ||
107 | - "transactions."); | ||
108 | - } | ||
109 | - | ||
110 | - @Override | ||
111 | - public CompletableFuture<Versioned<Collection<byte[]>>> removeAll(String key) { | ||
112 | - return submit(new UpdateAndGet(key, null, null, null)) | ||
113 | - .whenComplete((result, e) -> throwIfLocked(result.status())) | ||
114 | - .thenApply(result -> result.oldValue()); | ||
115 | } | 112 | } |
116 | 113 | ||
117 | @Override | 114 | @Override |
118 | - public CompletableFuture<Boolean> putAll(String key, Iterable<? extends byte[]> values) { | 115 | + public CompletableFuture< |
119 | - throw new UnsupportedOperationException("This operation cannot be " + | 116 | + Versioned<Collection<? extends byte[]>>> removeAll(String key) { |
120 | - "used without support for " + | 117 | + return submit(new RemoveAll(key, null)); |
121 | - "transactions."); | ||
122 | } | 118 | } |
123 | 119 | ||
124 | @Override | 120 | @Override |
125 | - public CompletableFuture<Boolean> putAll(Multimap<? extends String, ? extends byte[]> multiMap) { | 121 | + public CompletableFuture<Boolean> putAll( |
126 | - throw new UnsupportedOperationException("This operation cannot be " + | 122 | + String key, Collection<? extends byte[]> values) { |
127 | - "used without support for " + | 123 | + return submit(new Put(key, values, null)); |
128 | - "transactions."); | ||
129 | } | 124 | } |
130 | 125 | ||
131 | @Override | 126 | @Override |
132 | - public CompletableFuture<Collection<byte[]>> replaceValues(String key, Iterable<byte[]> values) { | 127 | + public CompletableFuture< |
133 | - throw new UnsupportedOperationException("This operation cannot be " + | 128 | + Versioned<Collection<? extends byte[]>>> replaceValues( |
134 | - "used without support for " + | 129 | + String key, Collection<byte[]> values) { |
135 | - "transactions."); | 130 | + return submit(new Replace(key, values, null)); |
136 | } | 131 | } |
137 | 132 | ||
138 | @Override | 133 | @Override |
139 | public CompletableFuture<Void> clear() { | 134 | public CompletableFuture<Void> clear() { |
140 | - return submit(new AsyncConsistentMultimapCommands.Clear()); | 135 | + return submit(new Clear()); |
141 | } | 136 | } |
142 | 137 | ||
143 | @Override | 138 | @Override |
144 | - public CompletableFuture<Collection<byte[]>> get(String key) { | 139 | + public CompletableFuture< |
145 | - return submit(new Get()); | 140 | + Versioned<Collection<? extends byte[]>>> get(String key) { |
141 | + return submit(new Get(key)); | ||
146 | } | 142 | } |
147 | 143 | ||
148 | @Override | 144 | @Override |
... | @@ -156,7 +152,7 @@ public class AsyncConsistentSetMultimap | ... | @@ -156,7 +152,7 @@ public class AsyncConsistentSetMultimap |
156 | } | 152 | } |
157 | 153 | ||
158 | @Override | 154 | @Override |
159 | - public CompletableFuture<Collection<byte[]>> values() { | 155 | + public CompletableFuture<Multiset<byte[]>> values() { |
160 | return submit(new Values()); | 156 | return submit(new Values()); |
161 | } | 157 | } |
162 | 158 | ||
... | @@ -182,7 +178,9 @@ public class AsyncConsistentSetMultimap | ... | @@ -182,7 +178,9 @@ public class AsyncConsistentSetMultimap |
182 | */ | 178 | */ |
183 | private void throwIfLocked(MapEntryUpdateResult.Status status) { | 179 | private void throwIfLocked(MapEntryUpdateResult.Status status) { |
184 | if (status == MapEntryUpdateResult.Status.WRITE_LOCK) { | 180 | if (status == MapEntryUpdateResult.Status.WRITE_LOCK) { |
185 | - throw new ConcurrentModificationException("Cannot update map: Another transaction in progress"); | 181 | + throw new ConcurrentModificationException("Cannot update map: " + |
182 | + "Another transaction " + | ||
183 | + "in progress"); | ||
186 | } | 184 | } |
187 | } | 185 | } |
188 | } | 186 | } | ... | ... |
1 | +/* | ||
2 | + * Copyright 2016 Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import io.atomix.catalyst.serializer.SerializableTypeResolver; | ||
20 | +import io.atomix.copycat.client.CopycatClient; | ||
21 | +import io.atomix.resource.ResourceFactory; | ||
22 | +import io.atomix.resource.ResourceStateMachine; | ||
23 | + | ||
24 | +import java.util.Properties; | ||
25 | + | ||
26 | +/** | ||
27 | + * {@link AsyncConsistentSetMultimap} resource factory. | ||
28 | + */ | ||
29 | +public class AsyncConsistentSetMultimapFactory implements | ||
30 | + ResourceFactory<AsyncConsistentSetMultimap> { | ||
31 | + @Override | ||
32 | + public SerializableTypeResolver createSerializableTypeResolver() { | ||
33 | + return new AsyncConsistentMultimapCommands.TypeResolver(); | ||
34 | + } | ||
35 | + | ||
36 | + @Override | ||
37 | + public ResourceStateMachine createStateMachine(Properties config) { | ||
38 | + return new AsyncConsistentSetMultimapState(config); | ||
39 | + } | ||
40 | + | ||
41 | + @Override | ||
42 | + public AsyncConsistentSetMultimap createInstance(CopycatClient client, | ||
43 | + Properties properties) { | ||
44 | + return new AsyncConsistentSetMultimap(client, properties); | ||
45 | + } | ||
46 | +} |
1 | +/* | ||
2 | + * Copyright 2016 Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import com.google.common.base.Preconditions; | ||
20 | +import com.google.common.collect.HashMultimap; | ||
21 | +import com.google.common.collect.HashMultiset; | ||
22 | +import com.google.common.collect.Lists; | ||
23 | +import com.google.common.collect.Maps; | ||
24 | +import com.google.common.collect.Multiset; | ||
25 | +import com.google.common.collect.Sets; | ||
26 | +import io.atomix.copycat.server.Commit; | ||
27 | +import io.atomix.copycat.server.Snapshottable; | ||
28 | +import io.atomix.copycat.server.StateMachineExecutor; | ||
29 | +import io.atomix.copycat.server.session.ServerSession; | ||
30 | +import io.atomix.copycat.server.session.SessionListener; | ||
31 | +import io.atomix.copycat.server.storage.snapshot.SnapshotReader; | ||
32 | +import io.atomix.copycat.server.storage.snapshot.SnapshotWriter; | ||
33 | +import io.atomix.resource.ResourceStateMachine; | ||
34 | +import org.onlab.util.CountDownCompleter; | ||
35 | +import org.onlab.util.Match; | ||
36 | +import org.onosproject.store.service.Versioned; | ||
37 | +import org.slf4j.Logger; | ||
38 | + | ||
39 | +import java.util.Arrays; | ||
40 | +import java.util.Collection; | ||
41 | +import java.util.Comparator; | ||
42 | +import java.util.EnumSet; | ||
43 | +import java.util.Map; | ||
44 | +import java.util.Properties; | ||
45 | +import java.util.Set; | ||
46 | +import java.util.TreeMap; | ||
47 | +import java.util.concurrent.atomic.AtomicLong; | ||
48 | +import java.util.function.BiConsumer; | ||
49 | +import java.util.function.BinaryOperator; | ||
50 | +import java.util.function.Function; | ||
51 | +import java.util.function.Supplier; | ||
52 | +import java.util.stream.Collector; | ||
53 | +import java.util.stream.Collectors; | ||
54 | + | ||
55 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Clear; | ||
56 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.ContainsEntry; | ||
57 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.ContainsKey; | ||
58 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.ContainsValue; | ||
59 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Entries; | ||
60 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Get; | ||
61 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.IsEmpty; | ||
62 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.KeySet; | ||
63 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Keys; | ||
64 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.MultiRemove; | ||
65 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.MultimapCommand; | ||
66 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Put; | ||
67 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.RemoveAll; | ||
68 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Replace; | ||
69 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Size; | ||
70 | +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.Values; | ||
71 | +import static org.slf4j.LoggerFactory.getLogger; | ||
72 | + | ||
73 | +/** | ||
74 | + * State Machine for {@link AsyncConsistentSetMultimap} resource. | ||
75 | + */ | ||
76 | +public class AsyncConsistentSetMultimapState extends ResourceStateMachine | ||
77 | + implements SessionListener, Snapshottable { | ||
78 | + | ||
79 | + private final Logger log = getLogger(getClass()); | ||
80 | + private final AtomicLong globalVersion = new AtomicLong(1); | ||
81 | + //TODO Add listener map here | ||
82 | + private final Map<String, MapEntryValue> backingMap = Maps.newHashMap(); | ||
83 | + | ||
84 | + public AsyncConsistentSetMultimapState(Properties properties) { | ||
85 | + super(properties); | ||
86 | + } | ||
87 | + | ||
88 | + @Override | ||
89 | + public void snapshot(SnapshotWriter writer) { | ||
90 | + } | ||
91 | + | ||
92 | + @Override | ||
93 | + public void install(SnapshotReader reader) { | ||
94 | + } | ||
95 | + | ||
96 | + @Override | ||
97 | + protected void configure(StateMachineExecutor executor) { | ||
98 | + executor.register(Size.class, this::size); | ||
99 | + executor.register(IsEmpty.class, this::isEmpty); | ||
100 | + executor.register(ContainsKey.class, this::containsKey); | ||
101 | + executor.register(ContainsValue.class, this::containsValue); | ||
102 | + executor.register(ContainsEntry.class, this::containsEntry); | ||
103 | + executor.register(Clear.class, this::clear); | ||
104 | + executor.register(KeySet.class, this::keySet); | ||
105 | + executor.register(Keys.class, this::keys); | ||
106 | + executor.register(Values.class, this::values); | ||
107 | + executor.register(Entries.class, this::entries); | ||
108 | + executor.register(Get.class, this::get); | ||
109 | + executor.register(RemoveAll.class, this::removeAll); | ||
110 | + executor.register(MultiRemove.class, this::multiRemove); | ||
111 | + executor.register(Put.class, this::put); | ||
112 | + executor.register(Replace.class, this::replace); | ||
113 | + } | ||
114 | + | ||
115 | + @Override | ||
116 | + public void delete() { | ||
117 | + super.delete(); | ||
118 | + } | ||
119 | + | ||
120 | + /** | ||
121 | + * Handles a Size commit. | ||
122 | + * | ||
123 | + * @param commit Size commit | ||
124 | + * @return number of unique key value pairs in the multimap | ||
125 | + */ | ||
126 | + protected int size(Commit<? extends Size> commit) { | ||
127 | + try { | ||
128 | + return backingMap.values() | ||
129 | + .stream() | ||
130 | + .map(valueCollection -> valueCollection.values().size()) | ||
131 | + .collect(Collectors.summingInt(size -> size)); | ||
132 | + } finally { | ||
133 | + commit.close(); | ||
134 | + } | ||
135 | + } | ||
136 | + | ||
137 | + /** | ||
138 | + * Handles an IsEmpty commit. | ||
139 | + * | ||
140 | + * @param commit IsEmpty commit | ||
141 | + * @return true if the multimap contains no key-value pairs, else false | ||
142 | + */ | ||
143 | + protected boolean isEmpty(Commit<? extends IsEmpty> commit) { | ||
144 | + try { | ||
145 | + return backingMap.isEmpty(); | ||
146 | + } finally { | ||
147 | + commit.close(); | ||
148 | + } | ||
149 | + } | ||
150 | + | ||
151 | + /** | ||
152 | + * Handles a contains key commit. | ||
153 | + * | ||
154 | + * @param commit ContainsKey commit | ||
155 | + * @return returns true if the key is in the multimap, else false | ||
156 | + */ | ||
157 | + protected boolean containsKey(Commit<? extends ContainsKey> commit) { | ||
158 | + try { | ||
159 | + return backingMap.containsKey(commit.operation().key()); | ||
160 | + } finally { | ||
161 | + commit.close(); | ||
162 | + } | ||
163 | + } | ||
164 | + | ||
165 | + /** | ||
166 | + * Handles a ContainsValue commit. | ||
167 | + * | ||
168 | + * @param commit ContainsValue commit | ||
169 | + * @return true if the value is in the multimap, else false | ||
170 | + */ | ||
171 | + protected boolean containsValue(Commit<? extends ContainsValue> commit) { | ||
172 | + try { | ||
173 | + Match<byte[]> match = Match.ifValue(commit.operation().value()); | ||
174 | + return backingMap | ||
175 | + .values() | ||
176 | + .stream() | ||
177 | + .anyMatch(valueList -> | ||
178 | + valueList | ||
179 | + .values() | ||
180 | + .stream() | ||
181 | + .anyMatch(byteValue -> | ||
182 | + match.matches(byteValue))); | ||
183 | + } finally { | ||
184 | + commit.close(); | ||
185 | + } | ||
186 | + } | ||
187 | + | ||
188 | + /** | ||
189 | + * Handles a ContainsEntry commit. | ||
190 | + * | ||
191 | + * @param commit ContainsEntry commit | ||
192 | + * @return true if the key-value pair exists, else false | ||
193 | + */ | ||
194 | + protected boolean containsEntry(Commit<? extends ContainsEntry> commit) { | ||
195 | + try { | ||
196 | + MapEntryValue entryValue = | ||
197 | + backingMap.get(commit.operation().key()); | ||
198 | + if (entryValue == null) { | ||
199 | + return false; | ||
200 | + } else { | ||
201 | + Match valueMatch = Match.ifValue(commit.operation().value()); | ||
202 | + return entryValue | ||
203 | + .values() | ||
204 | + .stream() | ||
205 | + .anyMatch(byteValue -> valueMatch.matches(byteValue)); | ||
206 | + } | ||
207 | + } finally { | ||
208 | + commit.close(); | ||
209 | + } | ||
210 | + } | ||
211 | + | ||
212 | + /** | ||
213 | + * Handles a Clear commit. | ||
214 | + * | ||
215 | + * @param commit Clear commit | ||
216 | + */ | ||
217 | + protected void clear(Commit<? extends Clear> commit) { | ||
218 | + try { | ||
219 | + backingMap.clear(); | ||
220 | + } finally { | ||
221 | + commit.close(); | ||
222 | + } | ||
223 | + } | ||
224 | + | ||
225 | + /** | ||
226 | + * Handles a KeySet commit. | ||
227 | + * | ||
228 | + * @param commit KeySet commit | ||
229 | + * @return a set of all keys in the multimap | ||
230 | + */ | ||
231 | + protected Set<String> keySet(Commit<? extends KeySet> commit) { | ||
232 | + try { | ||
233 | + return backingMap.keySet(); | ||
234 | + } finally { | ||
235 | + commit.close(); | ||
236 | + } | ||
237 | + } | ||
238 | + | ||
239 | + /** | ||
240 | + * Handles a Keys commit. | ||
241 | + * | ||
242 | + * @param commit Keys commit | ||
243 | + * @return a multiset of keys with each key included an equal number of | ||
244 | + * times to the total key-value pairs in which that key participates | ||
245 | + */ | ||
246 | + protected Multiset<String> keys(Commit<? extends Keys> commit) { | ||
247 | + try { | ||
248 | + Multiset keys = HashMultiset.create(); | ||
249 | + backingMap.forEach((key, mapEntryValue) -> { | ||
250 | + keys.add(key, mapEntryValue.values().size()); | ||
251 | + }); | ||
252 | + return keys; | ||
253 | + } finally { | ||
254 | + commit.close(); | ||
255 | + } | ||
256 | + } | ||
257 | + | ||
258 | + /** | ||
259 | + * Handles a Values commit. | ||
260 | + * | ||
261 | + * @param commit Values commit | ||
262 | + * @return the set of values in the multimap with duplicates included | ||
263 | + */ | ||
264 | + protected Multiset<byte[]> values(Commit<? extends Values> commit) { | ||
265 | + try { | ||
266 | + return backingMap | ||
267 | + .values() | ||
268 | + .stream() | ||
269 | + .collect(new HashMultisetValueCollector()); | ||
270 | + } finally { | ||
271 | + commit.close(); | ||
272 | + } | ||
273 | + } | ||
274 | + | ||
275 | + /** | ||
276 | + * Handles an Entries commit. | ||
277 | + * | ||
278 | + * @param commit Entries commit | ||
279 | + * @return a set of all key-value pairs in the multimap | ||
280 | + */ | ||
281 | + protected Collection<Map.Entry<String, byte[]>> entries( | ||
282 | + Commit<? extends Entries> commit) { | ||
283 | + try { | ||
284 | + return backingMap | ||
285 | + .entrySet() | ||
286 | + .stream() | ||
287 | + .collect(new EntrySetCollector()); | ||
288 | + } finally { | ||
289 | + commit.close(); | ||
290 | + } | ||
291 | + } | ||
292 | + | ||
293 | + /** | ||
294 | + * Handles a Get commit. | ||
295 | + * | ||
296 | + * @param commit Get commit | ||
297 | + * @return the collection of values associated with the key or an empty | ||
298 | + * list if none exist | ||
299 | + */ | ||
300 | + protected Versioned<Collection<? extends byte[]>> get( | ||
301 | + Commit<? extends Get> commit) { | ||
302 | + try { | ||
303 | + MapEntryValue mapEntryValue = backingMap.get(commit.operation().key()); | ||
304 | + return toVersioned(backingMap.get(commit.operation().key())); | ||
305 | + } finally { | ||
306 | + commit.close(); | ||
307 | + } | ||
308 | + } | ||
309 | + | ||
310 | + /** | ||
311 | + * Handles a removeAll commit, and returns the previous mapping. | ||
312 | + * | ||
313 | + * @param commit removeAll commit | ||
314 | + * @return collection of removed values | ||
315 | + */ | ||
316 | + protected Versioned<Collection<? extends byte[]>> removeAll( | ||
317 | + Commit<? extends RemoveAll> commit) { | ||
318 | + if (!backingMap.containsKey(commit.operation().key())) { | ||
319 | + commit.close(); | ||
320 | + return new Versioned<>(Sets.newHashSet(), -1); | ||
321 | + } else { | ||
322 | + return backingMap.get(commit.operation().key()).addCommit(commit); | ||
323 | + } | ||
324 | + } | ||
325 | + | ||
326 | + /** | ||
327 | + * Handles a multiRemove commit, returns true if the remove results in any | ||
328 | + * change. | ||
329 | + * @param commit multiRemove commit | ||
330 | + * @return true if any change results, else false | ||
331 | + */ | ||
332 | + protected boolean multiRemove(Commit<? extends MultiRemove> commit) { | ||
333 | + if (!backingMap.containsKey(commit.operation().key())) { | ||
334 | + commit.close(); | ||
335 | + return false; | ||
336 | + } else { | ||
337 | + return (backingMap | ||
338 | + .get(commit.operation().key()) | ||
339 | + .addCommit(commit)) != null; | ||
340 | + } | ||
341 | + } | ||
342 | + | ||
343 | + /** | ||
344 | + * Handles a put commit, returns true if any change results from this | ||
345 | + * commit. | ||
346 | + * @param commit a put commit | ||
347 | + * @return true if this commit results in a change, else false | ||
348 | + */ | ||
349 | + protected boolean put(Commit<? extends Put> commit) { | ||
350 | + if (commit.operation().values().isEmpty()) { | ||
351 | + return false; | ||
352 | + } | ||
353 | + if (!backingMap.containsKey(commit.operation().key())) { | ||
354 | + backingMap.put(commit.operation().key(), | ||
355 | + new NonTransactionalCommit(1)); | ||
356 | + } | ||
357 | + return backingMap | ||
358 | + .get(commit.operation().key()) | ||
359 | + .addCommit(commit) != null; | ||
360 | + } | ||
361 | + | ||
362 | + protected Versioned<Collection<? extends byte[]>> replace( | ||
363 | + Commit<? extends Replace> commit) { | ||
364 | + if (!backingMap.containsKey(commit.operation().key())) { | ||
365 | + backingMap.put(commit.operation().key(), | ||
366 | + new NonTransactionalCommit(1)); | ||
367 | + } | ||
368 | + return backingMap.get(commit.operation().key()).addCommit(commit); | ||
369 | + } | ||
370 | + | ||
371 | + @Override | ||
372 | + public void register(ServerSession session) { | ||
373 | + super.register(session); | ||
374 | + } | ||
375 | + | ||
376 | + @Override | ||
377 | + public void unregister(ServerSession session) { | ||
378 | + super.unregister(session); | ||
379 | + } | ||
380 | + | ||
381 | + @Override | ||
382 | + public void expire(ServerSession session) { | ||
383 | + super.expire(session); | ||
384 | + } | ||
385 | + | ||
386 | + @Override | ||
387 | + public void close(ServerSession session) { | ||
388 | + super.close(session); | ||
389 | + } | ||
390 | + | ||
391 | + private interface MapEntryValue { | ||
392 | + | ||
393 | + /** | ||
394 | + * Returns the list of raw {@code byte[]'s}. | ||
395 | + * | ||
396 | + * @return list of raw values | ||
397 | + */ | ||
398 | + Collection<? extends byte[]> values(); | ||
399 | + | ||
400 | + /** | ||
401 | + * Returns the version of the value. | ||
402 | + * | ||
403 | + * @return version | ||
404 | + */ | ||
405 | + long version(); | ||
406 | + | ||
407 | + /** | ||
408 | + * Discards the value by invoke appropriate clean up actions. | ||
409 | + */ | ||
410 | + void discard(); | ||
411 | + | ||
412 | + /** | ||
413 | + * Add a new commit and modifies the set of values accordingly. | ||
414 | + * In the case of a replace or removeAll it returns the set of removed | ||
415 | + * values. In the case of put or multiRemove it returns null for no | ||
416 | + * change and a set of the added or removed values respectively if a | ||
417 | + * change resulted. | ||
418 | + * | ||
419 | + * @param commit the commit to be added | ||
420 | + */ | ||
421 | + Versioned<Collection<? extends byte[]>> addCommit( | ||
422 | + Commit<? extends MultimapCommand> commit); | ||
423 | + } | ||
424 | + | ||
425 | + private class NonTransactionalCommit implements MapEntryValue { | ||
426 | + private long version; | ||
427 | + private final TreeMap<byte[], CountDownCompleter<Commit>> | ||
428 | + valueCountdownMap = Maps.newTreeMap(new ByteArrayComparator()); | ||
429 | + /*This is a mapping of commits that added values to the commits | ||
430 | + * removing those values, they will not be circular because keys will | ||
431 | + * be exclusively Put and Replace commits and values will be exclusively | ||
432 | + * Multiremove commits, each time a Put or replace is removed it should | ||
433 | + * as part of closing go through and countdown each of the remove | ||
434 | + * commits depending on it.*/ | ||
435 | + private final HashMultimap<Commit, CountDownCompleter<Commit>> | ||
436 | + additiveToRemovalCommits = HashMultimap.create(); | ||
437 | + | ||
438 | + public NonTransactionalCommit( | ||
439 | + long version) { | ||
440 | + //Set the version to current it will only be updated once this is | ||
441 | + // populated | ||
442 | + this.version = globalVersion.get(); | ||
443 | + } | ||
444 | + | ||
445 | + @Override | ||
446 | + public Collection<? extends byte[]> values() { | ||
447 | + return valueCountdownMap.keySet(); | ||
448 | + } | ||
449 | + | ||
450 | + @Override | ||
451 | + public long version() { | ||
452 | + return version; | ||
453 | + } | ||
454 | + | ||
455 | + @Override | ||
456 | + public void discard() { | ||
457 | + valueCountdownMap.values().forEach(completer -> | ||
458 | + completer.object().close()); | ||
459 | + } | ||
460 | + | ||
461 | + @Override | ||
462 | + public Versioned<Collection<? extends byte[]>> addCommit( | ||
463 | + Commit<? extends MultimapCommand> commit) { | ||
464 | + Preconditions.checkNotNull(commit); | ||
465 | + Preconditions.checkNotNull(commit.operation()); | ||
466 | + Versioned<Collection<? extends byte[]>> retVersion; | ||
467 | + | ||
468 | + if (commit.operation() instanceof Put) { | ||
469 | + //Using a treeset here sanitizes the input, removing duplicates | ||
470 | + Set<byte[]> valuesToAdd = | ||
471 | + Sets.newTreeSet(new ByteArrayComparator()); | ||
472 | + ((Put) commit.operation()).values().forEach(value -> { | ||
473 | + if (!valueCountdownMap.containsKey(value)) { | ||
474 | + valuesToAdd.add(value); | ||
475 | + } | ||
476 | + }); | ||
477 | + if (valuesToAdd.isEmpty()) { | ||
478 | + //Do not increment or add the commit if no change resulted | ||
479 | +// TODO fairly sure the below case is unreachable but | ||
480 | +// TODO need to make sure | ||
481 | +// if (valueCountdownMap.isEmpty()) { | ||
482 | +// backingMap.remove(((Put) commit.operation()).key()); | ||
483 | +// } | ||
484 | + commit.close(); | ||
485 | + return null; | ||
486 | + } | ||
487 | + //When all values from a commit have been removed decrement all | ||
488 | + //removal commits relying on it and remove itself from the | ||
489 | + //mapping of additive commits to the commits removing the | ||
490 | + //values it added. (Only multiremoves will be dependent) | ||
491 | + CountDownCompleter<Commit> completer = | ||
492 | + new CountDownCompleter<>(commit, valuesToAdd.size(), | ||
493 | + c -> { | ||
494 | + if (additiveToRemovalCommits.containsKey(c)) { | ||
495 | + additiveToRemovalCommits. | ||
496 | + get(c). | ||
497 | + forEach(countdown -> | ||
498 | + countdown.countDown()); | ||
499 | + additiveToRemovalCommits.removeAll(c); | ||
500 | + } | ||
501 | + c.close(); | ||
502 | + }); | ||
503 | + retVersion = new Versioned<>(valuesToAdd, version); | ||
504 | + valuesToAdd.forEach(value -> valueCountdownMap.put(value, | ||
505 | + completer)); | ||
506 | + version++; | ||
507 | + return retVersion; | ||
508 | + | ||
509 | + } else if (commit.operation() instanceof Replace) { | ||
510 | + //Will this work?? Need to check before check-in! | ||
511 | + Set<byte[]> removedValues = Sets.newHashSet(); | ||
512 | + removedValues.addAll(valueCountdownMap.keySet()); | ||
513 | + retVersion = new Versioned<>(removedValues, version); | ||
514 | + valueCountdownMap.values().forEach(countdown -> | ||
515 | + countdown.countDown()); | ||
516 | + valueCountdownMap.clear(); | ||
517 | + Set<byte[]> valuesToAdd = | ||
518 | + Sets.newTreeSet(new ByteArrayComparator()); | ||
519 | + ((Replace) commit.operation()).values().forEach(value -> { | ||
520 | + valuesToAdd.add(value); | ||
521 | + }); | ||
522 | + if (valuesToAdd.isEmpty()) { | ||
523 | + version = globalVersion.incrementAndGet(); | ||
524 | + backingMap.remove(((Replace) commit.operation()).key()); | ||
525 | + //Order is important here, the commit must be closed last | ||
526 | + //(or minimally after all uses) | ||
527 | + commit.close(); | ||
528 | + return retVersion; | ||
529 | + } | ||
530 | + CountDownCompleter<Commit> completer = | ||
531 | + new CountDownCompleter<>(commit, valuesToAdd.size(), | ||
532 | + c -> { | ||
533 | + if (additiveToRemovalCommits | ||
534 | + .containsKey(c)) { | ||
535 | + additiveToRemovalCommits. | ||
536 | + get(c). | ||
537 | + forEach(countdown -> | ||
538 | + countdown.countDown()); | ||
539 | + additiveToRemovalCommits. | ||
540 | + removeAll(c); | ||
541 | + } | ||
542 | + c.close(); | ||
543 | + }); | ||
544 | + valuesToAdd.forEach(value -> | ||
545 | + valueCountdownMap.put(value, completer)); | ||
546 | + version = globalVersion.incrementAndGet(); | ||
547 | + return retVersion; | ||
548 | + | ||
549 | + } else if (commit.operation() instanceof RemoveAll) { | ||
550 | + Set<byte[]> removed = Sets.newHashSet(); | ||
551 | + //We can assume here that values only appear once and so we | ||
552 | + //do not need to sanitize the return for duplicates. | ||
553 | + removed.addAll(valueCountdownMap.keySet()); | ||
554 | + retVersion = new Versioned<>(removed, version); | ||
555 | + valueCountdownMap.values().forEach(countdown -> | ||
556 | + countdown.countDown()); | ||
557 | + valueCountdownMap.clear(); | ||
558 | + //In the case of a removeAll all commits will be removed and | ||
559 | + //unlike the multiRemove case we do not need to consider | ||
560 | + //dependencies among additive and removal commits. | ||
561 | + | ||
562 | + //Save the key for use after the commit is closed | ||
563 | + String key = ((RemoveAll) commit.operation()).key(); | ||
564 | + commit.close(); | ||
565 | + version = globalVersion.incrementAndGet(); | ||
566 | + backingMap.remove(key); | ||
567 | + return retVersion; | ||
568 | + | ||
569 | + } else if (commit.operation() instanceof MultiRemove) { | ||
570 | + //Must first calculate how many commits the removal depends on. | ||
571 | + //At this time we also sanitize the removal set by adding to a | ||
572 | + //set with proper handling of byte[] equality. | ||
573 | + Set<byte[]> removed = Sets.newHashSet(); | ||
574 | + Set<Commit> commitsRemovedFrom = Sets.newHashSet(); | ||
575 | + ((MultiRemove) commit.operation()).values().forEach(value -> { | ||
576 | + if (valueCountdownMap.containsKey(value)) { | ||
577 | + removed.add(value); | ||
578 | + commitsRemovedFrom | ||
579 | + .add(valueCountdownMap.get(value).object()); | ||
580 | + } | ||
581 | + }); | ||
582 | + //If there is nothing to be removed no action should be taken. | ||
583 | + if (removed.isEmpty()) { | ||
584 | + //Do not increment or add the commit if no change resulted | ||
585 | + commit.close(); | ||
586 | + return null; | ||
587 | + } | ||
588 | + //When all additive commits this depends on are closed this can | ||
589 | + //be closed as well. | ||
590 | + CountDownCompleter<Commit> completer = | ||
591 | + new CountDownCompleter<>(commit, | ||
592 | + commitsRemovedFrom.size(), | ||
593 | + c -> c.close()); | ||
594 | + commitsRemovedFrom.forEach(commitRemovedFrom -> { | ||
595 | + additiveToRemovalCommits.put(commitRemovedFrom, completer); | ||
596 | + }); | ||
597 | + //Save key in case countdown results in closing the commit. | ||
598 | + String removedKey = ((MultiRemove) commit.operation()).key(); | ||
599 | + removed.forEach(removedValue -> { | ||
600 | + valueCountdownMap.remove(removedValue).countDown(); | ||
601 | + }); | ||
602 | + //The version is updated locally as well as globally even if | ||
603 | + //this object will be removed from the map in case any other | ||
604 | + //party still holds a reference to this object. | ||
605 | + retVersion = new Versioned<>(removed, version); | ||
606 | + version = globalVersion.incrementAndGet(); | ||
607 | + if (valueCountdownMap.isEmpty()) { | ||
608 | + backingMap | ||
609 | + .remove(removedKey); | ||
610 | + } | ||
611 | + return retVersion; | ||
612 | + | ||
613 | + } else { | ||
614 | + throw new IllegalArgumentException(); | ||
615 | + } | ||
616 | + } | ||
617 | + } | ||
618 | + | ||
619 | + /** | ||
620 | + * A collector that creates MapEntryValues and creates a multiset of all | ||
621 | + * values in the map an equal number of times to the number of sets in | ||
622 | + * which they participate. | ||
623 | + */ | ||
624 | + private class HashMultisetValueCollector implements | ||
625 | + Collector<MapEntryValue, | ||
626 | + HashMultiset<byte[]>, | ||
627 | + HashMultiset<byte[]>> { | ||
628 | + private HashMultiset<byte[]> multiset = null; | ||
629 | + | ||
630 | + @Override | ||
631 | + public Supplier<HashMultiset<byte[]>> supplier() { | ||
632 | + return new Supplier<HashMultiset<byte[]>>() { | ||
633 | + @Override | ||
634 | + public HashMultiset<byte[]> get() { | ||
635 | + if (multiset == null) { | ||
636 | + multiset = HashMultiset.create(); | ||
637 | + } | ||
638 | + return multiset; | ||
639 | + } | ||
640 | + }; | ||
641 | + } | ||
642 | + | ||
643 | + @Override | ||
644 | + public BiConsumer<HashMultiset<byte[]>, MapEntryValue> accumulator() { | ||
645 | + return (multiset, mapEntryValue) -> | ||
646 | + multiset.addAll(mapEntryValue.values()); | ||
647 | + } | ||
648 | + | ||
649 | + @Override | ||
650 | + public BinaryOperator<HashMultiset<byte[]>> combiner() { | ||
651 | + return (setOne, setTwo) -> { | ||
652 | + setOne.addAll(setTwo); | ||
653 | + return setOne; | ||
654 | + }; | ||
655 | + } | ||
656 | + | ||
657 | + @Override | ||
658 | + public Function<HashMultiset<byte[]>, | ||
659 | + HashMultiset<byte[]>> finisher() { | ||
660 | + return (unused) -> multiset; | ||
661 | + } | ||
662 | + | ||
663 | + @Override | ||
664 | + public Set<Characteristics> characteristics() { | ||
665 | + return EnumSet.of(Characteristics.UNORDERED); | ||
666 | + } | ||
667 | + } | ||
668 | + | ||
669 | + /** | ||
670 | + * A collector that creates Entries of {@code <String, MapEntryValue>} and | ||
671 | + * creates a set of entries all key value pairs in the map. | ||
672 | + */ | ||
673 | + private class EntrySetCollector implements | ||
674 | + Collector<Map.Entry<String, MapEntryValue>, | ||
675 | + Set<Map.Entry<String, byte[]>>, | ||
676 | + Set<Map.Entry<String, byte[]>>> { | ||
677 | + private Set<Map.Entry<String, byte[]>> set = null; | ||
678 | + | ||
679 | + @Override | ||
680 | + public Supplier<Set<Map.Entry<String, byte[]>>> supplier() { | ||
681 | + return new Supplier<Set<Map.Entry<String, byte[]>>>() { | ||
682 | + @Override | ||
683 | + public Set<Map.Entry<String, byte[]>> get() { | ||
684 | + if (set == null) { | ||
685 | + set = Sets.newHashSet(); | ||
686 | + } | ||
687 | + return set; | ||
688 | + } | ||
689 | + }; | ||
690 | + } | ||
691 | + | ||
692 | + @Override | ||
693 | + public BiConsumer<Set<Map.Entry<String, byte[]>>, | ||
694 | + Map.Entry<String, MapEntryValue>> accumulator() { | ||
695 | + return (set, entry) -> { | ||
696 | + entry | ||
697 | + .getValue() | ||
698 | + .values() | ||
699 | + .forEach(byteValue -> | ||
700 | + set.add(Maps.immutableEntry(entry.getKey(), | ||
701 | + byteValue))); | ||
702 | + }; | ||
703 | + } | ||
704 | + | ||
705 | + @Override | ||
706 | + public BinaryOperator<Set<Map.Entry<String, byte[]>>> combiner() { | ||
707 | + return (setOne, setTwo) -> { | ||
708 | + setOne.addAll(setTwo); | ||
709 | + return setOne; | ||
710 | + }; | ||
711 | + } | ||
712 | + | ||
713 | + @Override | ||
714 | + public Function<Set<Map.Entry<String, byte[]>>, | ||
715 | + Set<Map.Entry<String, byte[]>>> finisher() { | ||
716 | + return (unused) -> set; | ||
717 | + } | ||
718 | + | ||
719 | + @Override | ||
720 | + public Set<Characteristics> characteristics() { | ||
721 | + return EnumSet.of(Characteristics.UNORDERED); | ||
722 | + } | ||
723 | + } | ||
724 | + /** | ||
725 | + * Utility for turning a {@code MapEntryValue} to {@code Versioned}. | ||
726 | + * @param value map entry value | ||
727 | + * @return versioned instance or an empty list versioned -1 if argument is | ||
728 | + * null | ||
729 | + */ | ||
730 | + private Versioned<Collection<? extends byte[]>> toVersioned( | ||
731 | + MapEntryValue value) { | ||
732 | + return value == null ? new Versioned<>(Lists.newArrayList(), -1) : | ||
733 | + new Versioned<>(value.values(), | ||
734 | + value.version()); | ||
735 | + } | ||
736 | + | ||
737 | + private class ByteArrayComparator implements Comparator<byte[]> { | ||
738 | + | ||
739 | + @Override | ||
740 | + public int compare(byte[] o1, byte[] o2) { | ||
741 | + if (Arrays.equals(o1, o2)) { | ||
742 | + return 0; | ||
743 | + } else { | ||
744 | + for (int i = 0; i < o1.length && i < o2.length; i++) { | ||
745 | + if (o1[i] < o2[i]) { | ||
746 | + return -1; | ||
747 | + } else if (o1[i] > o2[i]) { | ||
748 | + return 1; | ||
749 | + } | ||
750 | + } | ||
751 | + return o1.length > o2.length ? 1 : -1; | ||
752 | + } | ||
753 | + } | ||
754 | + } | ||
755 | +} |
1 | +/* | ||
2 | + * Copyright 2016 Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import com.google.common.collect.Lists; | ||
20 | +import com.google.common.collect.Multiset; | ||
21 | +import com.google.common.collect.TreeMultiset; | ||
22 | +import com.google.common.io.Files; | ||
23 | +import io.atomix.catalyst.transport.Address; | ||
24 | +import io.atomix.catalyst.transport.LocalTransport; | ||
25 | +import io.atomix.copycat.server.CopycatServer; | ||
26 | +import io.atomix.copycat.server.storage.Storage; | ||
27 | +import io.atomix.copycat.server.storage.StorageLevel; | ||
28 | +import io.atomix.manager.state.ResourceManagerState; | ||
29 | +import io.atomix.resource.ResourceType; | ||
30 | +import org.apache.commons.collections.keyvalue.DefaultMapEntry; | ||
31 | +import org.junit.Ignore; | ||
32 | +import org.junit.Test; | ||
33 | +import org.onlab.util.Tools; | ||
34 | + | ||
35 | +import java.io.File; | ||
36 | +import java.time.Duration; | ||
37 | +import java.util.Arrays; | ||
38 | +import java.util.Collection; | ||
39 | +import java.util.Comparator; | ||
40 | +import java.util.List; | ||
41 | +import java.util.Map; | ||
42 | + | ||
43 | +import static org.junit.Assert.assertEquals; | ||
44 | +import static org.junit.Assert.assertFalse; | ||
45 | +import static org.junit.Assert.assertTrue; | ||
46 | + | ||
47 | +/** | ||
48 | + * Tests the {@link AsyncConsistentSetMultimap}. | ||
49 | + */ | ||
50 | +public class AsyncConsistentSetMultimapTest extends AtomixTestBase { | ||
51 | + private final File testDir = Files.createTempDir(); | ||
52 | + private final String keyOne = "hello"; | ||
53 | + private final String keyTwo = "goodbye"; | ||
54 | + private final String keyThree = "foo"; | ||
55 | + private final String keyFour = "bar"; | ||
56 | + private final byte[] valueOne = Tools.getBytesUtf8(keyOne); | ||
57 | + private final byte[] valueTwo = Tools.getBytesUtf8(keyTwo); | ||
58 | + private final byte[] valueThree = Tools.getBytesUtf8(keyThree); | ||
59 | + private final byte[] valueFour = Tools.getBytesUtf8(keyFour); | ||
60 | + private final List<String> allKeys = Lists.newArrayList(keyOne, keyTwo, | ||
61 | + keyThree, keyFour); | ||
62 | + private final List<byte[]> allValues = Lists.newArrayList(valueOne, | ||
63 | + valueTwo, | ||
64 | + valueThree, | ||
65 | + valueFour); | ||
66 | + | ||
67 | + @Override | ||
68 | + protected ResourceType resourceType() { | ||
69 | + return new ResourceType(AsyncConsistentSetMultimap.class); | ||
70 | + } | ||
71 | + | ||
72 | + /** | ||
73 | + * Test that size behaves correctly (This includes testing of the empty | ||
74 | + * check). | ||
75 | + */ | ||
76 | + @Ignore | ||
77 | + @Test | ||
78 | + public void testSize() throws Throwable { | ||
79 | + clearTests(); | ||
80 | + AsyncConsistentSetMultimap map = createResource(3); | ||
81 | + //Simplest operation case | ||
82 | + map.isEmpty().thenAccept(result -> assertTrue(result)); | ||
83 | + map.put(keyOne, valueOne). | ||
84 | + thenAccept(result -> assertTrue(result)).join(); | ||
85 | + map.isEmpty().thenAccept(result -> assertFalse(result)); | ||
86 | + map.size().thenAccept(result -> assertEquals(1, (int) result)) | ||
87 | + .join(); | ||
88 | + //Make sure sizing is dependent on values not keys | ||
89 | + map.put(keyOne, valueTwo). | ||
90 | + thenAccept(result -> assertTrue(result)).join(); | ||
91 | + map.size().thenAccept(result -> assertEquals(2, (int) result)) | ||
92 | + .join(); | ||
93 | + //Ensure that double adding has no effect | ||
94 | + map.put(keyOne, valueOne). | ||
95 | + thenAccept(result -> assertFalse(result)).join(); | ||
96 | + map.size().thenAccept(result -> assertEquals(2, (int) result)) | ||
97 | + .join(); | ||
98 | + //Check handling for multiple keys | ||
99 | + map.put(keyTwo, valueOne) | ||
100 | + .thenAccept(result -> assertTrue(result)).join(); | ||
101 | + map.put(keyTwo, valueTwo) | ||
102 | + .thenAccept(result -> assertTrue(result)).join(); | ||
103 | + map.size().thenAccept(result -> assertEquals(4, (int) result)) | ||
104 | + .join(); | ||
105 | + //Check size with removal | ||
106 | + map.remove(keyOne, valueOne). | ||
107 | + thenAccept(result -> assertTrue(result)).join(); | ||
108 | + map.size().thenAccept(result -> assertEquals(3, (int) result)) | ||
109 | + .join(); | ||
110 | + //Check behavior under remove of non-existant key | ||
111 | + map.remove(keyOne, valueOne). | ||
112 | + thenAccept(result -> assertFalse(result)).join(); | ||
113 | + map.size().thenAccept(result -> assertEquals(3, (int) result)) | ||
114 | + .join(); | ||
115 | + //Check clearing the entirety of the map | ||
116 | + map.clear().join(); | ||
117 | + map.size().thenAccept(result -> assertEquals(0, (int) result)) | ||
118 | + .join(); | ||
119 | + map.isEmpty().thenAccept(result -> assertTrue(result)); | ||
120 | + | ||
121 | + map.destroy().join(); | ||
122 | + clearTests(); | ||
123 | + } | ||
124 | + | ||
125 | + /** | ||
126 | + * Contains tests for value, key and entry. | ||
127 | + */ | ||
128 | + @Ignore | ||
129 | + @Test | ||
130 | + public void containsTest() throws Throwable { | ||
131 | + clearTests(); | ||
132 | + AsyncConsistentSetMultimap map = createResource(3); | ||
133 | + | ||
134 | + //Populate the maps | ||
135 | + allKeys.forEach(key -> { | ||
136 | + map.putAll(key, allValues) | ||
137 | + .thenAccept(result -> assertTrue(result)).join(); | ||
138 | + }); | ||
139 | + map.size().thenAccept(result -> assertEquals(16, (int) result)).join(); | ||
140 | + | ||
141 | + //Test key contains positive results | ||
142 | + allKeys.forEach(key -> { | ||
143 | + map.containsKey(key) | ||
144 | + .thenAccept(result -> assertTrue(result)).join(); | ||
145 | + }); | ||
146 | + | ||
147 | + //Test value contains positive results | ||
148 | + allValues.forEach(value -> { | ||
149 | + map.containsValue(value) | ||
150 | + .thenAccept(result -> assertTrue(result)).join(); | ||
151 | + }); | ||
152 | + | ||
153 | + //Test contains entry for all possible entries | ||
154 | + allKeys.forEach(key -> { | ||
155 | + allValues.forEach(value -> { | ||
156 | + map.containsEntry(key, value) | ||
157 | + .thenAccept(result -> assertTrue(result)).join(); | ||
158 | + }); | ||
159 | + }); | ||
160 | + | ||
161 | + //Test behavior after removals | ||
162 | + allValues.forEach(value -> { | ||
163 | + final String[] removedKey = new String[1]; | ||
164 | + allKeys.forEach(key -> { | ||
165 | + map.remove(key, value) | ||
166 | + .thenAccept(result -> assertTrue(result)).join(); | ||
167 | + map.containsEntry(key, value) | ||
168 | + .thenAccept(result -> assertFalse(result)).join(); | ||
169 | + removedKey[0] = key; | ||
170 | + }); | ||
171 | + //Check that contains key works properly for removed keys | ||
172 | + map.containsKey(removedKey[0]) | ||
173 | + .thenAccept(result -> assertFalse(result)); | ||
174 | + }); | ||
175 | + | ||
176 | + //Check that contains value works correctly for removed values | ||
177 | + allValues.forEach(value -> { | ||
178 | + map.containsValue(value) | ||
179 | + .thenAccept(result -> assertFalse(result)).join(); | ||
180 | + }); | ||
181 | + | ||
182 | + map.destroy().join(); | ||
183 | + clearTests(); | ||
184 | + } | ||
185 | + | ||
186 | + /** | ||
187 | + * Contains tests for put, putAll, remove, removeAll and replace. | ||
188 | + * @throws Exception | ||
189 | + */ | ||
190 | + @Ignore | ||
191 | + @Test | ||
192 | + public void addAndRemoveTest() throws Exception { | ||
193 | + clearTests(); | ||
194 | + AsyncConsistentSetMultimap map = createResource(3); | ||
195 | + | ||
196 | + //Test single put | ||
197 | + allKeys.forEach(key -> { | ||
198 | + //Value should actually be added here | ||
199 | + allValues.forEach(value -> { | ||
200 | + map.put(key, value) | ||
201 | + .thenAccept(result -> assertTrue(result)).join(); | ||
202 | + //Duplicate values should be ignored here | ||
203 | + map.put(key, value) | ||
204 | + .thenAccept(result -> assertFalse(result)).join(); | ||
205 | + }); | ||
206 | + }); | ||
207 | + | ||
208 | + //Test single remove | ||
209 | + allKeys.forEach(key -> { | ||
210 | + //Value should actually be added here | ||
211 | + allValues.forEach(value -> { | ||
212 | + map.remove(key, value) | ||
213 | + .thenAccept(result -> assertTrue(result)).join(); | ||
214 | + //Duplicate values should be ignored here | ||
215 | + map.remove(key, value) | ||
216 | + .thenAccept(result -> assertFalse(result)).join(); | ||
217 | + }); | ||
218 | + }); | ||
219 | + | ||
220 | + map.isEmpty().thenAccept(result -> assertTrue(result)).join(); | ||
221 | + | ||
222 | + //Test multi put | ||
223 | + allKeys.forEach(key -> { | ||
224 | + map.putAll(key, Lists.newArrayList(allValues.subList(0, 2))) | ||
225 | + .thenAccept(result -> assertTrue(result)).join(); | ||
226 | + map.putAll(key, Lists.newArrayList(allValues.subList(0, 2))) | ||
227 | + .thenAccept(result -> assertFalse(result)).join(); | ||
228 | + map.putAll(key, Lists.newArrayList(allValues.subList(2, 4))) | ||
229 | + .thenAccept(result -> assertTrue(result)).join(); | ||
230 | + map.putAll(key, Lists.newArrayList(allValues.subList(2, 4))) | ||
231 | + .thenAccept(result -> assertFalse(result)).join(); | ||
232 | + | ||
233 | + }); | ||
234 | + | ||
235 | + //Test multi remove | ||
236 | + allKeys.forEach(key -> { | ||
237 | + //Split the lists to test how multiRemove can work piecewise | ||
238 | + map.removeAll(key, Lists.newArrayList(allValues.subList(0, 2))) | ||
239 | + .thenAccept(result -> assertTrue(result)).join(); | ||
240 | + map.removeAll(key, Lists.newArrayList(allValues.subList(0, 2))) | ||
241 | + .thenAccept(result -> assertFalse(result)).join(); | ||
242 | + map.removeAll(key, Lists.newArrayList(allValues.subList(2, 4))) | ||
243 | + .thenAccept(result -> assertTrue(result)).join(); | ||
244 | + map.removeAll(key, Lists.newArrayList(allValues.subList(2, 4))) | ||
245 | + .thenAccept(result -> assertFalse(result)).join(); | ||
246 | + }); | ||
247 | + | ||
248 | + map.isEmpty().thenAccept(result -> assertTrue(result)).join(); | ||
249 | + | ||
250 | + //Repopulate for next test | ||
251 | + allKeys.forEach(key -> { | ||
252 | + map.putAll(key, allValues) | ||
253 | + .thenAccept(result -> assertTrue(result)).join(); | ||
254 | + }); | ||
255 | + | ||
256 | + map.size().thenAccept(result -> assertEquals(16, (int) result)).join(); | ||
257 | + | ||
258 | + //Test removeAll of entire entry | ||
259 | + allKeys.forEach(key -> { | ||
260 | + map.removeAll(key).thenAccept(result -> { | ||
261 | + assertTrue( | ||
262 | + byteArrayCollectionIsEqual(allValues, result.value())); | ||
263 | + }).join(); | ||
264 | + map.removeAll(key).thenAccept(result -> { | ||
265 | + assertFalse( | ||
266 | + byteArrayCollectionIsEqual(allValues, result.value())); | ||
267 | + }).join(); | ||
268 | + }); | ||
269 | + | ||
270 | + map.isEmpty().thenAccept(result -> assertTrue(result)).join(); | ||
271 | + | ||
272 | + //Repopulate for next test | ||
273 | + allKeys.forEach(key -> { | ||
274 | + map.putAll(key, allValues) | ||
275 | + .thenAccept(result -> assertTrue(result)).join(); | ||
276 | + }); | ||
277 | + | ||
278 | + map.size().thenAccept(result -> assertEquals(16, (int) result)).join(); | ||
279 | + | ||
280 | + allKeys.forEach(key -> { | ||
281 | + map.replaceValues(key, allValues) | ||
282 | + .thenAccept(result -> | ||
283 | + assertTrue(byteArrayCollectionIsEqual(allValues, | ||
284 | + result.value()))) | ||
285 | + .join(); | ||
286 | + map.replaceValues(key, Lists.newArrayList()) | ||
287 | + .thenAccept(result -> | ||
288 | + assertTrue(byteArrayCollectionIsEqual(allValues, | ||
289 | + result.value()))) | ||
290 | + .join(); | ||
291 | + map.replaceValues(key, allValues) | ||
292 | + .thenAccept(result -> | ||
293 | + assertTrue(result.value().isEmpty())) | ||
294 | + .join(); | ||
295 | + }); | ||
296 | + | ||
297 | + | ||
298 | + //Test replacements of partial sets | ||
299 | + map.size().thenAccept(result -> assertEquals(16, (int) result)).join(); | ||
300 | + | ||
301 | + allKeys.forEach(key -> { | ||
302 | + map.remove(key, valueOne) | ||
303 | + .thenAccept(result -> | ||
304 | + assertTrue(result)).join(); | ||
305 | + map.replaceValues(key, Lists.newArrayList()) | ||
306 | + .thenAccept(result -> | ||
307 | + assertTrue(byteArrayCollectionIsEqual( | ||
308 | + Lists.newArrayList(valueTwo, valueThree, | ||
309 | + valueFour), | ||
310 | + result.value()))) | ||
311 | + .join(); | ||
312 | + map.replaceValues(key, allValues) | ||
313 | + .thenAccept(result -> | ||
314 | + assertTrue(result.value().isEmpty())) | ||
315 | + .join(); | ||
316 | + }); | ||
317 | + | ||
318 | + map.destroy().join(); | ||
319 | + clearTests(); | ||
320 | + } | ||
321 | + | ||
322 | + /** | ||
323 | + * Tests the get, keySet, keys, values, and entries implementations as well | ||
324 | + * as a trivial test of the asMap functionality (throws error) | ||
325 | + * @throws Exception | ||
326 | + */ | ||
327 | + @Ignore | ||
328 | + @Test | ||
329 | + public void testAccessors() throws Exception { | ||
330 | + clearTests(); | ||
331 | + AsyncConsistentSetMultimap map = createResource(3); | ||
332 | + | ||
333 | + //Populate for full map behavior tests | ||
334 | + allKeys.forEach(key -> { | ||
335 | + map.putAll(key, allValues) | ||
336 | + .thenAccept(result -> assertTrue(result)).join(); | ||
337 | + }); | ||
338 | + | ||
339 | + map.size().thenAccept(result -> assertEquals(16, (int) result)).join(); | ||
340 | + | ||
341 | + allKeys.forEach(key -> { | ||
342 | + map.get(key).thenAccept(result -> { | ||
343 | + assertTrue(byteArrayCollectionIsEqual(allValues, | ||
344 | + result.value())); | ||
345 | + }).join(); | ||
346 | + }); | ||
347 | + | ||
348 | + //Test that the key set is correct | ||
349 | + map.keySet() | ||
350 | + .thenAccept(result -> | ||
351 | + assertTrue(stringArrayCollectionIsEqual(allKeys, | ||
352 | + result))) | ||
353 | + .join(); | ||
354 | + //Test that the correct set and occurrence of values are found in the | ||
355 | + //values result | ||
356 | + map.values().thenAccept(result -> { | ||
357 | + final Multiset<byte[]> set = TreeMultiset.create( | ||
358 | + new ByteArrayComparator()); | ||
359 | + for (int i = 0; i < 4; i++) { | ||
360 | + set.addAll(allValues); | ||
361 | + } | ||
362 | + assertEquals(16, result.size()); | ||
363 | + result.forEach(value -> assertTrue(set.remove(value))); | ||
364 | + assertTrue(set.isEmpty()); | ||
365 | + | ||
366 | + }).join(); | ||
367 | + | ||
368 | + //Test that keys returns the right result including the correct number | ||
369 | + //of each item | ||
370 | + map.keys().thenAccept(result -> { | ||
371 | + final Multiset<String> set = TreeMultiset.create(); | ||
372 | + for (int i = 0; i < 4; i++) { | ||
373 | + set.addAll(allKeys); | ||
374 | + } | ||
375 | + assertEquals(16, result.size()); | ||
376 | + result.forEach(value -> assertTrue(set.remove(value))); | ||
377 | + assertTrue(set.isEmpty()); | ||
378 | + | ||
379 | + }).join(); | ||
380 | + | ||
381 | + //Test that the right combination of key, value pairs are present | ||
382 | + map.entries().thenAccept(result -> { | ||
383 | + final Multiset<Map.Entry<String, byte[]>> set = | ||
384 | + TreeMultiset.create(new EntryComparator()); | ||
385 | + allKeys.forEach(key -> { | ||
386 | + allValues.forEach(value -> { | ||
387 | + set.add(new DefaultMapEntry(key, value)); | ||
388 | + }); | ||
389 | + }); | ||
390 | + assertEquals(16, result.size()); | ||
391 | + result.forEach(entry -> assertTrue(set.remove(entry))); | ||
392 | + assertTrue(set.isEmpty()); | ||
393 | + }).join(); | ||
394 | + | ||
395 | + | ||
396 | + //Testing for empty map behavior | ||
397 | + map.clear().join(); | ||
398 | + | ||
399 | + allKeys.forEach(key -> { | ||
400 | + map.get(key).thenAccept(result -> { | ||
401 | + assertTrue(result.value().isEmpty()); | ||
402 | + }).join(); | ||
403 | + }); | ||
404 | + | ||
405 | + map.keySet().thenAccept(result -> assertTrue(result.isEmpty())).join(); | ||
406 | + map.values().thenAccept(result -> assertTrue(result.isEmpty())).join(); | ||
407 | + map.keys().thenAccept(result -> assertTrue(result.isEmpty())).join(); | ||
408 | + map.entries() | ||
409 | + .thenAccept(result -> assertTrue(result.isEmpty())).join(); | ||
410 | + | ||
411 | + map.destroy(); | ||
412 | + clearTests(); | ||
413 | + } | ||
414 | + | ||
415 | + | ||
416 | + private AsyncConsistentSetMultimap createResource(int clusterSize) { | ||
417 | + try { | ||
418 | + createCopycatServers(clusterSize); | ||
419 | + AsyncConsistentSetMultimap map = createAtomixClient(). | ||
420 | + getResource("testMap", AsyncConsistentSetMultimap.class) | ||
421 | + .join(); | ||
422 | + return map; | ||
423 | + } catch (Throwable e) { | ||
424 | + throw new RuntimeException(e.toString()); | ||
425 | + } | ||
426 | + } | ||
427 | + | ||
428 | + @Override | ||
429 | + protected CopycatServer createCopycatServer(Address address) { | ||
430 | + CopycatServer server = CopycatServer.builder(address, members) | ||
431 | + .withTransport(new LocalTransport(registry)) | ||
432 | + .withStorage(Storage.builder() | ||
433 | + .withStorageLevel(StorageLevel.MEMORY) | ||
434 | + .withDirectory(testDir + "/" + address.port()) | ||
435 | + .build()) | ||
436 | + .withStateMachine(ResourceManagerState::new) | ||
437 | + .withSerializer(serializer.clone()) | ||
438 | + .withHeartbeatInterval(Duration.ofMillis(25)) | ||
439 | + .withElectionTimeout(Duration.ofMillis(50)) | ||
440 | + .withSessionTimeout(Duration.ofMillis(100)) | ||
441 | + .build(); | ||
442 | + copycatServers.add(server); | ||
443 | + return server; } | ||
444 | + | ||
445 | + /** | ||
446 | + * Returns two arrays contain the same set of elements, | ||
447 | + * regardless of order. | ||
448 | + * @param o1 first collection | ||
449 | + * @param o2 second collection | ||
450 | + * @return true if they contain the same elements | ||
451 | + */ | ||
452 | + private boolean byteArrayCollectionIsEqual( | ||
453 | + Collection<? extends byte[]> o1, Collection<? extends byte[]> o2) { | ||
454 | + if (o1 == null || o2 == null || o1.size() != o2.size()) { | ||
455 | + return false; | ||
456 | + } | ||
457 | + for (byte[] array1 : o1) { | ||
458 | + boolean matched = false; | ||
459 | + for (byte[] array2 : o2) { | ||
460 | + if (Arrays.equals(array1, array2)) { | ||
461 | + matched = true; | ||
462 | + break; | ||
463 | + } | ||
464 | + } | ||
465 | + if (!matched) { | ||
466 | + return false; | ||
467 | + } | ||
468 | + } | ||
469 | + return true; | ||
470 | + } | ||
471 | + | ||
472 | + /** | ||
473 | + * Compares two collections of strings returns true if they contain the | ||
474 | + * same strings, false otherwise. | ||
475 | + * @param s1 string collection one | ||
476 | + * @param s2 string collection two | ||
477 | + * @return true if the two sets contain the same strings | ||
478 | + */ | ||
479 | + private boolean stringArrayCollectionIsEqual( | ||
480 | + Collection<? extends String> s1, Collection<? extends String> s2) { | ||
481 | + if (s1 == null || s2 == null || s1.size() != s2.size()) { | ||
482 | + return false; | ||
483 | + } | ||
484 | + for (String string1 : s1) { | ||
485 | + boolean matched = false; | ||
486 | + for (String string2 : s2) { | ||
487 | + if (string1.equals(string2)) { | ||
488 | + matched = true; | ||
489 | + break; | ||
490 | + } | ||
491 | + } | ||
492 | + if (!matched) { | ||
493 | + return false; | ||
494 | + } | ||
495 | + } | ||
496 | + return true; | ||
497 | + } | ||
498 | + | ||
499 | + /** | ||
500 | + * Byte array comparator implementation. | ||
501 | + */ | ||
502 | + private class ByteArrayComparator implements Comparator<byte[]> { | ||
503 | + | ||
504 | + @Override | ||
505 | + public int compare(byte[] o1, byte[] o2) { | ||
506 | + if (Arrays.equals(o1, o2)) { | ||
507 | + return 0; | ||
508 | + } else { | ||
509 | + for (int i = 0; i < o1.length && i < o2.length; i++) { | ||
510 | + if (o1[i] < o2[i]) { | ||
511 | + return -1; | ||
512 | + } else if (o1[i] > o2[i]) { | ||
513 | + return 1; | ||
514 | + } | ||
515 | + } | ||
516 | + return o1.length > o2.length ? 1 : -1; | ||
517 | + } | ||
518 | + } | ||
519 | + } | ||
520 | + | ||
521 | + /** | ||
522 | + * Entry comparator, uses both key and value to determine equality, | ||
523 | + * for comparison falls back to the default string comparator. | ||
524 | + */ | ||
525 | + private class EntryComparator | ||
526 | + implements Comparator<Map.Entry<String, byte[]>> { | ||
527 | + | ||
528 | + @Override | ||
529 | + public int compare(Map.Entry<String, byte[]> o1, | ||
530 | + Map.Entry<String, byte[]> o2) { | ||
531 | + if (o1 == null || o1.getKey() == null || o2 == null || | ||
532 | + o2.getKey() == null) { | ||
533 | + throw new IllegalArgumentException(); | ||
534 | + } | ||
535 | + if (o1.getKey().equals(o2.getKey()) && | ||
536 | + Arrays.equals(o1.getValue(), o2.getValue())) { | ||
537 | + return 0; | ||
538 | + } else { | ||
539 | + return o1.getKey().compareTo(o2.getKey()); | ||
540 | + } | ||
541 | + } | ||
542 | + } | ||
543 | +} |
-
Please register or login to post a comment