=== modified file 'src/com/goldencode/p2j/directory/LockManagerTest.java'
--- old/src/com/goldencode/p2j/directory/LockManagerTest.java 2026-05-07 16:00:31 +0000
+++ new/src/com/goldencode/p2j/directory/LockManagerTest.java 2026-05-18 06:10:11 +0000
@@ -240,7 +240,7 @@
public static void main(String[] args)
{
//Create an instance of LockManager
- LockManager man = new LocalLockManager();
+ LockManager man = new RedisLockManager();
//Simplest tests:
System.out.println("Test set #1 (simple R/O locks)");
=== modified file 'src/com/goldencode/p2j/directory/RedisLockManager.java'
--- old/src/com/goldencode/p2j/directory/RedisLockManager.java 2026-05-17 13:54:49 +0000
+++ new/src/com/goldencode/p2j/directory/RedisLockManager.java 2026-05-18 06:11:13 +0000
@@ -1,178 +1,126 @@
-/*
-** Module : RedisLockManager.java
-** Abstract : Implementation of LockManager for Redis backend.
-**
-** Copyright (c) 2005-2026, Golden Code Development Corporation.
-**
-** -#- -I- --Date-- ----------------------------------Description----------------------------------
-** 001 ES 20260516 Created initial version.
-*/
-
-/*
-** This program is free software: you can redistribute it and/or modify
-** it under the terms of the GNU Affero General Public License as
-** published by the Free Software Foundation, either version 3 of the
-** License, or (at your option) any later version.
-**
-** This program is distributed in the hope that it will be useful,
-** but WITHOUT ANY WARRANTY; without even the implied warranty of
-** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-** GNU Affero General Public License for more details.
-**
-** You may find a copy of the GNU Affero GPL version 3 at the following
-** location: https://www.gnu.org/licenses/agpl-3.0.en.html
-**
-** Additional terms under GNU Affero GPL version 3 section 7:
-**
-** Under Section 7 of the GNU Affero GPL version 3, the following additional
-** terms apply to the works covered under the License. These additional terms
-** are non-permissive additional terms allowed under Section 7 of the GNU
-** Affero GPL version 3 and may not be removed by you.
-**
-** 0. Attribution Requirement.
-**
-** You must preserve all legal notices or author attributions in the covered
-** work or Appropriate Legal Notices displayed by works containing the covered
-** work. You may not remove from the covered work any author or developer
-** credit already included within the covered work.
-**
-** 1. No License To Use Trademarks.
-**
-** This license does not grant any license or rights to use the trademarks
-** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
-** of Golden Code Development Corporation. You are not authorized to use the
-** name Golden Code, FWD, or the names of any author or contributor, for
-** publicity purposes without written authorization.
-**
-** 2. No Misrepresentation of Affiliation.
-**
-** You may not represent yourself as Golden Code Development Corporation or FWD.
-**
-** You may not represent yourself for publicity purposes as associated with
-** Golden Code Development Corporation, FWD, or any author or contributor to
-** the covered work, without written authorization.
-**
-** 3. No Misrepresentation of Source or Origin.
-**
-** You may not represent the covered work as solely your work. All modified
-** versions of the covered work must be marked in a reasonable way to make it
-** clear that the modified work is not originating from Golden Code Development
-** Corporation or FWD. All modified versions must contain the notices of
-** attribution required in this license.
-*/
-
package com.goldencode.p2j.directory;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
-
-import org.redisson.api.RLock;
-import org.redisson.api.RReadWriteLock;
-import org.redisson.api.RedissonClient;
-
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
import com.goldencode.p2j.util.logging.CentralLogger;
-/**
- * A Redis-backed implementation of the {@link LockManager} interface.
- *
- * This manager uses Redisson to provide distributed, hierarchical locking for
- * directory tree nodes. It splits a node's path into segments and acquires
- * appropriate read/write locks ("traffic" and "batch" locks) along the path
- * to ensure cluster-wide consistency during concurrent operations.
- *
- */
public class RedisLockManager implements LockManager
{
- /** The Redisson client used to acquire distributed locks. */
- private RedissonClient redisson;
-
- /**
- * Constructs a new {@code RedisLockManager} and initializes the Redisson client
- * from the global {@link RedisConfig}.
- */
+ private static final CentralLogger LOG = CentralLogger.get(RedisLockManager.class);
+
+ private static final String JVM_ID = java.util.UUID.randomUUID().toString();
+ private static final String LOCK_PREFIX = "dir:lock:tree:";
+ private static final int SPIN_SLEEP_MS = 10;
+
+ private JedisPool jedisPool;
+
+ private static volatile String lockSha = null;
+ private static volatile String releaseSha = null;
+
+ private volatile boolean shutdown = false;
+
+ private static final String LUA_LOCK_SCRIPT =
+ "local targetType = ARGV[1]; " +
+ "local threadId = ARGV[2]; " +
+ "for i, key in ipairs(KEYS) do " +
+ "local lockKeys = redis.call('HKEYS', key); " +
+ "for j, tId in ipairs(lockKeys) do " +
+ "if tId ~= threadId then " +
+ "local lType = redis.call('HGET', key, tId); " +
+ "if lType == 'WX' or targetType == 'WX' then return \"0\"; end " +
+ "if targetType == 'RX' then return \"0\"; end " +
+ "if lType == 'RX' or lType == 'INTENT_RX' then " +
+ "if targetType ~= 'RO' then return \"0\"; end " +
+ "end " +
+ "end " +
+ "end " +
+ "end " +
+ "local intentType = (targetType == 'RX') and 'INTENT_RX' or 'INTENT'; " +
+ "for i, key in ipairs(KEYS) do " +
+ "local applyType = (i == #KEYS) and targetType or intentType; " +
+ "redis.call('HSET', key, threadId, applyType); " +
+ "end " +
+ "return \"1\";";
+
+ private static final String LUA_UNLOCK_SCRIPT =
+ "local threadId = ARGV[1]; " +
+ "for i, key in ipairs(KEYS) do " +
+ "redis.call('HDEL', key, threadId); " +
+ "end " +
+ "return \"1\";";
+
+ private static final String LUA_UPGRADE_SCRIPT =
+ "local threadId = ARGV[1]; " +
+ "local targetKey = KEYS[1]; " +
+ "local currentType = redis.call('HGET', targetKey, threadId); " +
+ "if currentType == 'RX' then " +
+ "if redis.call('HLEN', targetKey) > 1 then return \"0\"; end " +
+ "redis.call('HSET', targetKey, threadId, 'WX'); " +
+ "return \"1\"; " +
+ "end " +
+ "return \"0\";";
+
public RedisLockManager()
{
}
- /**
- * Acquires a distributed lock for a specific directory node and its ancestors.
- *
- * The method iterates through the path hierarchy (e.g., from root down to the target node),
- * acquiring sequential Read/Write locks. This prevents parent nodes from being modified
- * or deleted while child nodes are actively being accessed or modified.
- *
- *
- * @param id
- * The normalized string path of the node to lock (e.g., "parent/child").
- * @param type
- * The type of lock requested (e.g., {@code LOCK_RW}, {@code LOCK_RX}, {@code LOCK_WX}).
- *
- * @return An opaque lock reference object ({@link LockRef}) if the locks were successfully acquired,
- * or {@code null} if the acquisition failed or inputs were invalid.
- */
@Override
public Object enterLock(String id, int type)
{
- if (id == null || type == LOCK_WX)
- {
- return null;
- }
+ if (id == null || shutdown) return null;
id = IdUtils.normalize(id);
- String[] pathParts = splitPath(id);
- List acquiredLocks = new ArrayList<>();
-
- if (redisson == null)
+ List keys = buildHierarchyKeys(id);
+ String threadId = JVM_ID + ":" + Thread.currentThread().getId();
+ String typeStr = getTypeString(type);
+
+ List args = new ArrayList<>();
+ args.add(typeStr);
+ args.add(threadId);
+
+ if (jedisPool == null)
{
RedisConfig cfg = RedisConfig.getRedisConfig();
- redisson = cfg.getRedissonClient();
- }
-
- try
- {
- for (int i = 0; i < pathParts.length; i++)
- {
- String path = pathParts[i];
- boolean isTarget = (i == pathParts.length - 1);
-
- RReadWriteLock trafficLock = redisson.getReadWriteLock("dir:lock:traffic:" + path);
- RReadWriteLock batchLock = redisson.getReadWriteLock("dir:lock:batch:" + path);
-
- if (type == LOCK_RW || type == LOCK_RX)
- {
- boolean needsTrafficWrite = isTarget && (type == LOCK_RX);
- RLock tLock = needsTrafficWrite ? trafficLock.writeLock() : trafficLock.readLock();
-
- tLock.lockInterruptibly();
- acquiredLocks.add(tLock);
- }
-
- RLock bLock = batchLock.readLock();
- bLock.lockInterruptibly();
- acquiredLocks.add(bLock);
-
- if (isTarget)
- {
- return new LockRef(acquiredLocks, type, batchLock);
- }
- }
- }
- catch (InterruptedException e)
- {
- Thread.currentThread().interrupt();
- releaseAll(acquiredLocks);
- return null;
- }
-
+ jedisPool = cfg.getJedisPool();
+ }
+
+ while (!shutdown)
+ {
+ try (Jedis jedis = jedisPool.getResource())
+ {
+ if (lockSha == null || releaseSha == null)
+ {
+ lockSha = jedis.scriptLoad(LUA_LOCK_SCRIPT);
+ releaseSha = jedis.scriptLoad(LUA_UNLOCK_SCRIPT);
+ }
+
+ Object res = jedis.evalsha(lockSha, keys, args);
+ if (res != null && "1".equals(res.toString()))
+ {
+ return new LockRef(keys, type, threadId, this);
+ }
+ }
+ catch (Exception e)
+ {
+ LOG.warning("Transient error acquiring lock for path: " + id, e);
+ }
+
+ try
+ {
+ Thread.sleep(SPIN_SLEEP_MS);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ return null;
+ }
+ }
return null;
}
- /**
- * Releases all distributed locks associated with the provided lock reference.
- *
- * @param objLock
- * The lock reference object returned by {@link #enterLock(String, int)}.
- */
@Override
public void releaseLock(Object objLock)
{
@@ -182,15 +130,6 @@
}
}
- /**
- * Attempts to upgrade an existing read lock ({@code LOCK_RX}) to an exclusive
- * write lock ({@code LOCK_WX}).
- *
- * @param objLock
- * The lock reference object to upgrade.
- *
- * @return {@code true} if the lock was successfully upgraded, {@code false} otherwise.
- */
@Override
public boolean upgradeLock(Object objLock)
{
@@ -202,144 +141,120 @@
return ((LockRef) objLock).upgrade();
}
- /**
- * Gracefully shuts down the lock manager.
- * Currently, this method acts as a stub.
- */
@Override
public synchronized void shutdownManager()
{
- // this.shutdown = true;
- }
-
- /**
- * Releases a list of acquired locks asynchronously in reverse order of their acquisition.
- *
- * @param locks
- * The list of Redisson locks to release.
- */
- private void releaseAll(List locks)
- {
- for (int i = locks.size() - 1; i >= 0; i--)
+ this.shutdown = true;
+ if (jedisPool != null && !jedisPool.isClosed())
{
- RLock lock = locks.get(i);
- if (lock.isHeldByCurrentThread())
- {
- lock.unlockAsync();
- }
+ jedisPool.close();
}
}
- /**
- * Splits a hierarchical path string into an array of progressive path segments.
- * For example, "a/b/c" becomes {@code ["a", "a/b", "a/b/c"]}.
- *
- * @param id
- * The full path string to split.
- *
- * @return An array of progressively built path strings.
- */
- private String[] splitPath(String id)
+ private List buildHierarchyKeys(String id)
{
- if (id.isEmpty() || id.equals("/"))
+ if (id.isEmpty() || "/".equals(id))
{
- return new String[]{""};
+ return Collections.singletonList(LOCK_PREFIX + "");
}
String[] segments = id.split("/");
- String[] paths = new String[segments.length];
+ List keys = new ArrayList<>();
StringBuilder current = new StringBuilder();
- for (int i = 0; i < segments.length; i++)
+
+ for (String segment : segments)
{
- if (i > 0)
+ if (segment.isEmpty())
+ {
+ keys.add(LOCK_PREFIX + current.toString());
+ continue;
+ }
+
+ if (current.length() > 0)
{
current.append("/");
}
- current.append(segments[i]);
- paths[i] = current.toString();
+ current.append(segment);
+ keys.add(LOCK_PREFIX + current.toString());
}
- return paths;
- }
-
- /**
- * An internal wrapper class that tracks the state and collection of distributed
- * locks acquired during a lock operation.
- */
+ return keys;
+ }
+
+ private String getTypeString(int type)
+ {
+ switch (type)
+ {
+ case LOCK_RX: return "RX";
+ case LOCK_RW: return "RW";
+ case LOCK_WX: return "WX";
+ default: return "RO";
+ }
+ }
+
static class LockRef
{
- /** The sequential list of locks acquired along the node hierarchy. */
- private final List locks;
-
- /** The current logical lock type (e.g., LOCK_RX, LOCK_WX). */
+ private final List keys;
private int type;
-
- /** The specific read/write batch lock associated with the target node. */
- private RReadWriteLock batchLock;
+ private final String threadId;
+ private final RedisLockManager manager;
- /**
- * Creates a new lock reference.
- *
- * @param locks
- * The list of acquired locks.
- * @param type
- * The type of lock requested.
- * @param batchLock
- * The Redisson ReadWrite lock of the target node.
- */
- public LockRef(List locks, int type, RReadWriteLock batchLock)
+ public LockRef(List keys, int type, String threadId, RedisLockManager manager)
{
- this.locks = locks;
+ this.keys = keys;
this.type = type;
- this.batchLock = batchLock;
+ this.threadId = threadId;
+ this.manager = manager;
}
- /**
- * Upgrades a {@code LOCK_RX} (Read Exclusive) lock to a {@code LOCK_WX} (Write Exclusive) lock.
- * It releases the current read lock and blocks until the write lock can be acquired.
- *
- * @return {@code true} if upgraded successfully, {@code false} if the current lock
- * type does not support upgrading.
- */
public boolean upgrade()
{
- if (type == LOCK_RX)
+ if (this.type != LOCK_RX) return false;
+
+ List targetKeyList = Collections.singletonList(keys.get(keys.size() - 1));
+ List args = new ArrayList<>();
+ args.add(threadId);
+
+ while (!manager.shutdown)
{
+ try (Jedis jedis = manager.jedisPool.getResource())
+ {
+ Object res = jedis.eval(LUA_UPGRADE_SCRIPT, targetKeyList, args);
+ if (res != null && "1".equals(res.toString()))
+ {
+ this.type = LOCK_WX;
+ return true;
+ }
+ }
+ catch (Exception e)
+ {
+ LOG.warning("Exception during lock upgrade attempt", e);
+ }
+
try
{
- RLock rdLock = batchLock.readLock();
- RLock wxLock = batchLock.writeLock();
-
- rdLock.unlock();
- wxLock.lockInterruptibly();
-
- locks.add(wxLock);
- this.type = LOCK_WX;
- return true;
- }
+ Thread.sleep(SPIN_SLEEP_MS);
+ }
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
+ return false;
}
}
-
return false;
}
- /**
- * Synchronously releases all locks tracked by this reference in reverse order
- * of their acquisition, ensuring safe tear-down of hierarchical locks.
- */
public void release()
{
- for (int i = locks.size() - 1; i >= 0; i--)
- {
- RLock lock = locks.get(i);
- if (lock.isHeldByCurrentThread())
- {
- lock.unlock();
- }
+ try (Jedis jedis = manager.jedisPool.getResource())
+ {
+ List args = Collections.singletonList(threadId);
+ jedis.evalsha(releaseSha, keys, args);
+ }
+ catch (Exception e)
+ {
+ LOG.warning("Failed to clean up keys in Redis on release", e);
}
}
}
=== modified file 'src/com/goldencode/p2j/directory/RedisRemapper.java'
--- old/src/com/goldencode/p2j/directory/RedisRemapper.java 2026-05-17 13:54:49 +0000
+++ new/src/com/goldencode/p2j/directory/RedisRemapper.java 2026-05-18 06:10:11 +0000
@@ -101,7 +101,7 @@
private static final String KEY_ATTR = "dir:attr:%s:%s";
/** Global Redis key used for synchronization during initialization/load. */
- private static final String LOCK_KEY = "dir:lock:batch:";
+ private static final String LOCK_KEY = "dir:lock:tree:";
/** Redis key used to flag that the system is under maintenance (read-only mode). */
private static final String MAINTENANCE = "maintenance";
@@ -112,6 +112,24 @@
/** XML attribute name used to store the structural version/revision in the root element. */
private static final String VERSION_ATTR_NAME = "version";
+ /**
+ * Atomically checks if a load is in progress. If not, it wipes the DB,
+ * sets the lock, and returns 1. If someone else is loading, it returns 0.
+ */
+ private static final String LUA_FORCE_LOAD_INIT =
+ "local lockKey = KEYS[1]; " +
+ "local threadId = ARGV[1]; " +
+ "redis.call('FLUSHDB'); " +
+ "redis.call('HSET', lockKey, threadId, 'WX')" +
+ "return \"1\";";
+
+ private static final String LUA_RELEASE_LOAD_LOCK =
+ "local threadId = ARGV[1]; " +
+ "for i, key in ipairs(KEYS) do " +
+ "redis.call('HDEL', key, threadId); " +
+ "end " +
+ "return \"1\";";
+
/** JSON Mapper used to serialize/deserialize {@link Attribute} objects. */
private static final ObjectMapper json_mapper = new ObjectMapper();
@@ -272,30 +290,40 @@
throws Exception
{
if (!initialize)
- {
- RReadWriteLock lock = redissonClient.getReadWriteLock(LOCK_KEY);
+ {
+ String threadId = java.util.UUID.randomUUID().toString();
- try
+ try (Jedis jedis = jedisPool.getResource())
{
- lock.writeLock().lockInterruptibly();
+ Object res = jedis.eval(LUA_FORCE_LOAD_INIT,
+ Collections.singletonList(LOCK_KEY),
+ Collections.singletonList(threadId));
- try (Jedis jedis = jedisPool.getResource())
- {
- if (file == null || !file.exists() || !file.isFile())
+ try
+ {
+ if (res != null && "1".equals(res.toString()))
{
- return;
+ switch (res.toString())
+ {
+ case "1":
+ {
+ if (file == null || !file.exists() || !file.isFile())
+ {
+ return;
+ }
+
+ XmlRedisRemapperIO io = new XmlRedisRemapperIO(this, jedis);
+ io.load(new FileInputStream(file));
+ initialize = true;
+ }
+ }
}
-
- XmlRedisRemapperIO io = new XmlRedisRemapperIO(this, jedis);
- io.load(new FileInputStream(file));
- initialize = true;
}
- }
- finally
- {
- if (lock != null && lock.writeLock().isHeldByCurrentThread())
+ finally
{
- lock.writeLock().unlock();
+ jedis.eval(LUA_RELEASE_LOAD_LOCK,
+ Collections.singletonList(LOCK_KEY),
+ Collections.singletonList(threadId));
}
}
}