You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@geode.apache.org by GitBox <gi...@apache.org> on 2020/04/28 12:39:29 UTC
[GitHub] [geode] prettyClouds commented on a change in pull request #5009: Deltas
prettyClouds commented on a change in pull request #5009:
URL: https://github.com/apache/geode/pull/5009#discussion_r416573027
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/executor/set/DeltaSet.java
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.redis.internal.executor.set;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.geode.DataSerializable;
+import org.apache.geode.DataSerializer;
+import org.apache.geode.Delta;
+import org.apache.geode.InvalidDeltaException;
+import org.apache.geode.cache.Region;
+import org.apache.geode.redis.internal.ByteArrayWrapper;
+
+/**
+ * TODO: it is probably a bad idea for this class to implement Set.
+ * We want to be careful how other code interacts with these instances
+ * to make sure that no modifications are made that are not thread safe
+ * and that will always be stored in the region.
+ * Currently the only "correct" methods on this class are:
+ * members, delete, customAddAll, customRemoveAll, and the
+ * serialization methods.
+ */
+public class DeltaSet implements Delta, DataSerializable {
+
+ public static long sadd(Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key,
+ ArrayList<ByteArrayWrapper> membersToAdd) {
+ while (true) {
Review comment:
If we synchronize "lower" in the stack, can we avoid the complexity of looping and throwing exceptions. Can we somehow guarantee that on a given key, only one of these operations is happening simultaneously? Maybe an exclusive lock on the key?
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/executor/set/SaddFunction.java
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.redis.internal.executor.set;
+
+import java.util.ArrayList;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.execute.Function;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
+import org.apache.geode.redis.internal.ByteArrayWrapper;
+
+class SaddFunction implements Function {
Review comment:
When we first implemented this a few weeks back, Jens suggested having just one remote function that encapsulated all the Set commands, and the execute could just switch on some parameter. I don't have a strong feeling either way, but I can't remember Jens' rationale...just wanted to capture the thought here, before it gets lost.
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/executor/set/DeltaSet.java
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.redis.internal.executor.set;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.geode.DataSerializable;
+import org.apache.geode.DataSerializer;
+import org.apache.geode.Delta;
+import org.apache.geode.InvalidDeltaException;
+import org.apache.geode.cache.Region;
+import org.apache.geode.redis.internal.ByteArrayWrapper;
+
+/**
+ * TODO: it is probably a bad idea for this class to implement Set.
+ * We want to be careful how other code interacts with these instances
+ * to make sure that no modifications are made that are not thread safe
+ * and that will always be stored in the region.
+ * Currently the only "correct" methods on this class are:
+ * members, delete, customAddAll, customRemoveAll, and the
+ * serialization methods.
+ */
+public class DeltaSet implements Delta, DataSerializable {
+
+ public static long sadd(Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key,
+ ArrayList<ByteArrayWrapper> membersToAdd) {
+ while (true) {
+ DeltaSet deltaSet = region.get(key);
+ if (deltaSet == null) {
+ // create new set
+ if (region.putIfAbsent(key, new DeltaSet(membersToAdd)) == null) {
+ return membersToAdd.size();
+ } else {
+ // retry since another thread concurrently changed the region
+ }
+ } else {
+ // update existing value
+ try {
+ return deltaSet.saddInstance(membersToAdd, region, key);
+ } catch (DeltaSet.RetryDueToConcurrentModification ex) {
+ // retry since another thread concurrently changed the region
+ }
+ }
+ }
+ }
+
+ public static long srem(Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key,
+ ArrayList<ByteArrayWrapper> membersToRemove) {
+ while (true) {
+ DeltaSet deltaSet = region.get(key);
+ if (deltaSet == null) {
+ return 0L;
+ }
+ try {
+ return deltaSet.sremInstance(membersToRemove, region, key);
+ } catch (DeltaSet.RetryDueToConcurrentModification ex) {
+ // retry since another thread concurrently changed the region
+ }
+ }
+ }
+
+ public static boolean del(Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key) {
+ while (true) {
+ DeltaSet deltaSet = region.get(key);
+ if (deltaSet == null) {
+ return false;
+ }
+ if (deltaSet.delInstance(region, key)) {
+ return true;
+ } else {
+ // retry since another thread concurrently changed the region
+ }
+ }
+ }
+
+ public static Set<ByteArrayWrapper> members(Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key) {
+ DeltaSet deltaSet = region.get(key);
+ if (deltaSet != null) {
+ return deltaSet.members();
+ } else {
+ return Collections.emptySet();
+ }
+ }
+
+ public synchronized boolean contains(ByteArrayWrapper member) {
+ return members.contains(member);
+ }
+
+ public synchronized int size() {
+ return members.size();
+ }
+
+ private HashSet<ByteArrayWrapper> members;
+ private transient ArrayList<ByteArrayWrapper> deltas;
+ // true if deltas contains adds; false if removes
+ private transient boolean deltasAreAdds;
+
+ DeltaSet(Collection<ByteArrayWrapper> members) {
+ if (members instanceof HashSet) {
+ this.members = (HashSet<ByteArrayWrapper>) members;
+ } else {
+ this.members = new HashSet<>(members);
+ }
+ }
+
+ // for serialization
+ public DeltaSet() {}
+
+
+ // DELTA
+ @Override
+ public boolean hasDelta() {
+ return deltas != null;
+ }
+
+ @Override
+ public void toDelta(DataOutput out) throws IOException {
+ DataSerializer.writeBoolean(deltasAreAdds, out);
+ DataSerializer.writeArrayList(deltas, out);
+ }
+
+ @Override
+ public synchronized void fromDelta(DataInput in)
+ throws IOException, InvalidDeltaException {
+ boolean deltaAdds = DataSerializer.readBoolean(in);
+ try {
+ ArrayList<ByteArrayWrapper> deltas = DataSerializer.readArrayList(in);
+ if (deltas != null) {
+ if (deltaAdds) {
+ members.addAll(deltas);
+ } else {
+ members.removeAll(deltas);
+ }
+ }
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // DATA SERIALIZABLE
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeHashSet(members, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ members = DataSerializer.readHashSet(in);
+ }
+
+
+ /**
+ * @param membersToAdd members to add to this set; NOTE this list may by
+ * modified by this call
+ * @param region the region this instance is stored in
+ * @param key the name of the set to add to
+ * @return the number of members actually added
+ * @throws RetryDueToConcurrentModification if a concurrent modification is detected
+ */
+ private synchronized long saddInstance(ArrayList<ByteArrayWrapper> membersToAdd,
+ Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key) {
+ if (region.get(key) != this) {
+ throw new RetryDueToConcurrentModification();
+ }
+ membersToAdd.removeIf(memberToAdd -> !members.add(memberToAdd));
+ int membersAdded = membersToAdd.size();
+ if (membersAdded != 0) {
+ deltasAreAdds = true;
+ deltas = membersToAdd;
+ try {
+ region.put(key, this);
+ } finally {
+ deltas = null;
+ }
+ }
+ return membersAdded;
+ }
+
+ /**
+ * @param membersToRemove members to remove from this set; NOTE this list may by
+ * modified by this call
+ * @param region the region this instance is stored in
+ * @param key the name of the set to remove from
+ * @return the number of members actually removed
+ * @throws RetryDueToConcurrentModification if a concurrent modification is detected
+ */
+ private synchronized long sremInstance(ArrayList<ByteArrayWrapper> membersToRemove,
+ Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key) {
+ if (region.get(key) != this) {
+ throw new RetryDueToConcurrentModification();
+ }
+ membersToRemove.removeIf(memberToRemove -> !members.remove(memberToRemove));
+ int membersRemoved = membersToRemove.size();
+ if (membersRemoved != 0) {
+ deltasAreAdds = false;
+ deltas = membersToRemove;
+ try {
+ region.put(key, this);
+ } finally {
+ deltas = null;
+ }
+ }
+ return membersRemoved;
+ }
+
+ /**
+ * This exception is thrown if a modification fails because some other
+ * thread changed what is stored in the region.
+ */
+ static class RetryDueToConcurrentModification extends RuntimeException {
Review comment:
This exception name tells the User what to do, i.e. "retry", instead of just stating what happened and letting the user decide how to handle it.
Maybe just throwing a ConcurrentModificationException, or something like RegionUnexpectedModificationException
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/executor/set/SaddFunction.java
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.redis.internal.executor.set;
+
+import java.util.ArrayList;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.execute.Function;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
+import org.apache.geode.redis.internal.ByteArrayWrapper;
+
+class SaddFunction implements Function {
+
+ public static final String ID = "SADD_FUNCTION";
+
+ @SuppressWarnings("unchecked")
Review comment:
How do we get rid of these? Adding a type to the class declaration, like `... implements Function<Object>`?
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/executor/set/DeltaSet.java
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.redis.internal.executor.set;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.geode.DataSerializable;
+import org.apache.geode.DataSerializer;
+import org.apache.geode.Delta;
+import org.apache.geode.InvalidDeltaException;
+import org.apache.geode.cache.Region;
+import org.apache.geode.redis.internal.ByteArrayWrapper;
+
+/**
+ * TODO: it is probably a bad idea for this class to implement Set.
+ * We want to be careful how other code interacts with these instances
+ * to make sure that no modifications are made that are not thread safe
+ * and that will always be stored in the region.
+ * Currently the only "correct" methods on this class are:
+ * members, delete, customAddAll, customRemoveAll, and the
+ * serialization methods.
+ */
+public class DeltaSet implements Delta, DataSerializable {
+
+ public static long sadd(Region<ByteArrayWrapper, DeltaSet> region,
+ ByteArrayWrapper key,
+ ArrayList<ByteArrayWrapper> membersToAdd) {
+ while (true) {
Review comment:
We could even use some kind of custom read-write lock...sadds should be able to happen in parallel, and srems should be able to happen in parallel exclusive of sadds, and dels just have to be exclusive of everything.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org