Enhance service semaphore to support parameter-based locking

framework/service/dtd/services.xsd

@@ -355,6 +355,19 @@ under the License.
             </xs:simpleType>
         </xs:attribute>
         <xs:attribute name="optional" type="xs:boolean" default="false"/>
+        <xs:attribute name="include-in-lock" type="xs:boolean" default="false">
+            <xs:annotation>
+                <xs:documentation>
+                    Only relevant when the service defines a semaphore (wait or fail).
+                    If set to true the value of this attribute is included in the semaphore lock key:
+                    the values of all attributes flagged with include-in-lock are hashed and the hash is
+                    appended to the service name used to acquire the lock, so calls with different values
+                    can run concurrently while calls with the same values are still serialized.
+                    Can only be used on IN or INOUT attributes, ignored otherwise.
+                    An optional attribute can be flagged too, its value is then included as null when not passed.
+                </xs:documentation>
+            </xs:annotation>
+        </xs:attribute>
         <xs:attribute name="default-value" type="xs:string">
             <xs:annotation>
                 <xs:documentation>The value specified will be used for the attribute if no value is passed in.
@@ -436,6 +449,14 @@ under the License.
             </xs:simpleType>
         </xs:attribute>
         <xs:attribute name="optional" type="xs:boolean"/>
+        <xs:attribute name="include-in-lock" type="xs:boolean" default="false">
+            <xs:annotation>
+                <xs:documentation>
+                    See the documentation on the include-in-lock attribute of the "attribute" element.
+                    Useful to flag an attribute defined through auto-attributes.
+                </xs:documentation>
+            </xs:annotation>
+        </xs:attribute>
         <xs:attribute name="default-value" type="xs:string"/>
         <xs:attribute name="form-label" type="xs:string"/>
         <xs:attribute name="form-display" type="xs:boolean"/>

framework/service/src/main/java/org/apache/ofbiz/service/ModelParam.java

@@ -95,6 +95,8 @@ public class ModelParam implements Serializable {
 
     /** Is this Parameter set internally? */
     private boolean internal = false;
+    /** Is this Parameter value included in the service semaphore lock key? */
+    private boolean includeInLock = false;
     /** Children attributes*/
     private ArrayList<ModelParam> children = null;
 
@@ -122,6 +124,7 @@ public ModelParam(ModelParam param) {
         this.overrideFormDisplay = param.overrideFormDisplay;
         this.allowHtml = param.allowHtml;
         this.internal = param.internal;
+        this.includeInLock = param.includeInLock;
     }
 
     /**
@@ -484,6 +487,22 @@ public boolean isOptional() {
         return this.optional;
     }
 
+    /**
+     * Is include in lock boolean.
+     * @return the boolean
+     */
+    public boolean isIncludeInLock() {
+        return this.includeInLock;
+    }
+
+    /**
+     * Sets include in lock.
+     * @param includeInLock the include in lock
+     */
+    public void setIncludeInLock(boolean includeInLock) {
+        this.includeInLock = includeInLock;
+    }
+
     /**
      * Gets default value.
      * @return the default value

framework/service/src/main/java/org/apache/ofbiz/service/ModelService.java

@@ -1912,6 +1912,14 @@ public synchronized void interfaceUpdate(DispatchContext dctx) throws GenericSer
                         if (overrideParam.isOverrideOptional()) {
                             existingParam.setOptional(overrideParam.isOptional());
                         }
+                        if (overrideParam.isIncludeInLock()) {
+                            if (!existingParam.isIn()) {
+                                Debug.logError("Attribute [" + overrideParam.getName() + "] of service [" + this.name
+                                        + "] cannot be included in the semaphore lock key: only IN or INOUT attributes are allowed", MODULE);
+                            } else {
+                                existingParam.setIncludeInLock(true);
+                            }
+                        }
                         if (UtilValidate.isNotEmpty(overrideParam.getAllowHtml())) {
                             existingParam.setAllowHtml(overrideParam.getAllowHtml());
                         }

framework/service/src/main/java/org/apache/ofbiz/service/ModelServiceReader.java

@@ -529,6 +529,16 @@ private static ModelParam createAttrDef(Element attribute, ModelParam parentPara
             param.setDefaultValue(defValue.intern());
         }
 
+        // flag the attribute to include its value in the service semaphore lock key
+        if ("true".equalsIgnoreCase(attribute.getAttribute("include-in-lock"))) {
+            if (!param.isIn()) {
+                Debug.logError("Attribute [" + param.getName() + "] of service [" + service.getName()
+                        + "] cannot be included in the semaphore lock key: only IN or INOUT attributes are allowed", MODULE);
+            } else {
+                param.setIncludeInLock(true);
+            }
+        }
+
         // set the entity name to the default if not specified
         if (param.getEntityName().length() == 0) {
             param.setEntityName(service.getDefaultEntityName());
@@ -595,6 +605,15 @@ private static void createOverrideDefs(Element baseElement, ModelService service
                     param.setAllowHtml(UtilXml.checkEmpty(overrideElement.getAttribute("allow-html")).intern());
                 }
 
+                if ("true".equalsIgnoreCase(overrideElement.getAttribute("include-in-lock"))) {
+                    if (!param.isIn()) {
+                        Debug.logError("Attribute [" + param.getName() + "] of service [" + service.getName()
+                                + "] cannot be included in the semaphore lock key: only IN or INOUT attributes are allowed", MODULE);
+                    } else {
+                        param.setIncludeInLock(true);
+                    }
+                }
+
                 // default value
                 String defValue = overrideElement.getAttribute("default-value");
                 if (UtilValidate.isNotEmpty(defValue)) {

framework/service/src/main/java/org/apache/ofbiz/service/ServiceDispatcher.java

@@ -290,7 +290,7 @@ public Map<String, Object> runSync(String localName, ModelService modelService,
         try {
             // check for semaphore and acquire a lock
             if ("wait".equals(modelService.getSemaphore()) || "fail".equals(modelService.getSemaphore())) {
-                lock = new ServiceSemaphore(delegator, modelService);
+                lock = new ServiceSemaphore(delegator, modelService, params);
                 lock.acquire();
             }
 

framework/service/src/main/java/org/apache/ofbiz/service/semaphore/ServiceSemaphore.java

@@ -18,18 +18,26 @@
  *******************************************************************************/
 package org.apache.ofbiz.service.semaphore;
 
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.sql.Timestamp;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 import javax.transaction.Transaction;
 
 import org.apache.ofbiz.base.util.Debug;
+import org.apache.ofbiz.base.util.StringUtil;
 import org.apache.ofbiz.base.util.UtilDateTime;
 import org.apache.ofbiz.entity.Delegator;
 import org.apache.ofbiz.entity.GenericEntityException;
 import org.apache.ofbiz.entity.GenericValue;
 import org.apache.ofbiz.entity.transaction.GenericTransactionException;
 import org.apache.ofbiz.entity.transaction.TransactionUtil;
 import org.apache.ofbiz.entity.util.EntityQuery;
+import org.apache.ofbiz.service.ModelParam;
 import org.apache.ofbiz.service.ModelService;
 import org.apache.ofbiz.service.job.JobManager;
 
@@ -50,17 +58,50 @@ public final class ServiceSemaphore {
     private Delegator delegator;
     private GenericValue lock;
     private ModelService model;
+    private String lockName;
 
     private int wait = 0;
     private int mode;
     private Timestamp lockTime = null;
 
-    public ServiceSemaphore(Delegator delegator, ModelService model) {
+    public ServiceSemaphore(Delegator delegator, ModelService model, Map<String, ?> context) {
         this.delegator = delegator;
         this.mode = "wait".equals(model.getSemaphore()) ? SEMAPHORE_MODE_WAIT
                 : ("fail".equals(model.getSemaphore()) ? SEMAPHORE_MODE_FAIL : SEMAPHORE_MODE_NONE);
         this.model = model;
         this.lock = null;
+        this.lockName = makeLockName(model, context);
+    }
+
+    /**
+     * Build the semaphore lock name for a service call. When some service attributes are flagged
+     * with include-in-lock="true", their values are hashed and the hash is appended to the service
+     * name so that the lock scope is the combination of the service and those attribute values,
+     * while the ServiceSemaphore entity keeps its single field primary key.
+     * @param model the service model
+     * @param context the service call context
+     * @return the lock name stored in the ServiceSemaphore serviceName field
+     */
+    private static String makeLockName(ModelService model, Map<String, ?> context) {
+        List<ModelParam> lockParams = model.getInModelParamList().stream()
+                .filter(ModelParam::isIncludeInLock)
+                .collect(Collectors.toList());
+        if (lockParams.isEmpty()) {
+            return model.getName();
+        }
+        StringBuilder lockKey = new StringBuilder();
+        for (ModelParam lockParam : lockParams) {
+            Object value = context != null ? context.get(lockParam.getName()) : null;
+            lockKey.append(lockParam.getName()).append('=').append(value).append(';');
+        }
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            String hash = StringUtil.toHexString(digest.digest(lockKey.toString().getBytes(StandardCharsets.UTF_8)));
+            // the hash is truncated so the lock name fits in the 100 character serviceName field
+            return model.getName() + "#" + hash.substring(0, 32);
+        } catch (NoSuchAlgorithmException e) {
+            throw new IllegalStateException("SHA-256 message digest not available", e);
+        }
     }
 
     /**
@@ -101,7 +142,7 @@ public synchronized boolean release() {
     private void waitOrFail() throws SemaphoreWaitException, SemaphoreFailException {
         if (SEMAPHORE_MODE_FAIL == mode) {
             // fail
-            throw new SemaphoreFailException("Service [" + model.getName() + "] is locked");
+            throw new SemaphoreFailException("Service [" + lockName + "] is locked");
         } else if (SEMAPHORE_MODE_WAIT == mode) {
             // get the wait and sleep values
             long maxWaitCount = ((model.getSemaphoreWait() * 1000) / model.getSemaphoreSleep());
@@ -124,7 +165,7 @@ private void waitOrFail() throws SemaphoreWaitException, SemaphoreFailException
             }
             if (timedOut) {
                 double waitTimeSec = ((System.currentTimeMillis() - lockTime.getTime()) / 1000.0);
-                String errMsg = "Service [" + model.getName() + "] with wait semaphore exceeded wait timeout, waited ["
+                String errMsg = "Service [" + lockName + "] with wait semaphore exceeded wait timeout, waited ["
                         + waitTimeSec + "], wait started at " + lockTime;
                 throw new SemaphoreWaitException(errMsg);
             }
@@ -147,8 +188,8 @@ private boolean checkLockNeedToWait() throws SemaphoreFailException {
 
         try {
             if (EntityQuery.use(delegator).from("ServiceSemaphore")
-                    .where("serviceName", model.getName()).queryCount() == 0) {
-                semaphore = delegator.makeValue("ServiceSemaphore", "serviceName", model.getName(),
+                    .where("serviceName", lockName).queryCount() == 0) {
+                semaphore = delegator.makeValue("ServiceSemaphore", "serviceName", lockName,
                         "lockedByInstanceId", JobManager.INSTANCE_ID, "lockThread", threadName, "lockTime", lockTime);
 
                 // use the special method below so we can reuse the unique tx functions
@@ -196,7 +237,7 @@ private synchronized boolean dbWrite(GenericValue value, boolean delete) {
                 } else {
                     // Last check before inserting data in this transaction to avoid error log
                     isError = EntityQuery.use(delegator).from("ServiceSemaphore")
-                            .where("serviceName", model.getName()).queryCount() != 0;
+                            .where("serviceName", lockName).queryCount() != 0;
                     if (!isError) {
                         lock = value.create();
                     }