Project

General

Profile

lua_script_lockManager.patch

Eduard Soltan, 05/18/2026 02:12 AM

Download (22.5 KB)

View differences:

new/src/com/goldencode/p2j/directory/LockManagerTest.java 2026-05-18 06:10:11 +0000
240 240
   public static void main(String[] args)
241 241
   {
242 242
      //Create an instance of LockManager
243
      LockManager man = new LocalLockManager();
243
      LockManager man = new RedisLockManager();
244 244

  
245 245
      //Simplest tests:
246 246
      System.out.println("Test set #1 (simple R/O locks)");
new/src/com/goldencode/p2j/directory/RedisLockManager.java 2026-05-18 06:11:13 +0000
1
/*
2
** Module   : RedisLockManager.java
3
** Abstract : Implementation of LockManager for Redis backend.
4
**
5
** Copyright (c) 2005-2026, Golden Code Development Corporation.
6
**
7
** -#- -I- --Date-- ----------------------------------Description----------------------------------
8
** 001 ES 20260516 Created initial version.
9
*/
10

  
11
/*
12
** This program is free software: you can redistribute it and/or modify
13
** it under the terms of the GNU Affero General Public License as
14
** published by the Free Software Foundation, either version 3 of the
15
** License, or (at your option) any later version.
16
**
17
** This program is distributed in the hope that it will be useful,
18
** but WITHOUT ANY WARRANTY; without even the implied warranty of
19
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20
** GNU Affero General Public License for more details.
21
**
22
** You may find a copy of the GNU Affero GPL version 3 at the following
23
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
24
** 
25
** Additional terms under GNU Affero GPL version 3 section 7:
26
** 
27
**   Under Section 7 of the GNU Affero GPL version 3, the following additional
28
**   terms apply to the works covered under the License.  These additional terms
29
**   are non-permissive additional terms allowed under Section 7 of the GNU
30
**   Affero GPL version 3 and may not be removed by you.
31
** 
32
**   0. Attribution Requirement.
33
** 
34
**     You must preserve all legal notices or author attributions in the covered
35
**     work or Appropriate Legal Notices displayed by works containing the covered
36
**     work.  You may not remove from the covered work any author or developer
37
**     credit already included within the covered work.
38
** 
39
**   1. No License To Use Trademarks.
40
** 
41
**     This license does not grant any license or rights to use the trademarks
42
**     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
43
**     of Golden Code Development Corporation. You are not authorized to use the
44
**     name Golden Code, FWD, or the names of any author or contributor, for
45
**     publicity purposes without written authorization.
46
** 
47
**   2. No Misrepresentation of Affiliation.
48
** 
49
**     You may not represent yourself as Golden Code Development Corporation or FWD.
50
** 
51
**     You may not represent yourself for publicity purposes as associated with
52
**     Golden Code Development Corporation, FWD, or any author or contributor to
53
**     the covered work, without written authorization.
54
** 
55
**   3. No Misrepresentation of Source or Origin.
56
** 
57
**     You may not represent the covered work as solely your work.  All modified
58
**     versions of the covered work must be marked in a reasonable way to make it
59
**     clear that the modified work is not originating from Golden Code Development
60
**     Corporation or FWD.  All modified versions must contain the notices of
61
**     attribution required in this license.
62
*/
63

  
64 1
package com.goldencode.p2j.directory;
65 2

  
66 3
import java.util.ArrayList;
4
import java.util.Collections;
67 5
import java.util.List;
68

  
69
import org.redisson.api.RLock;
70
import org.redisson.api.RReadWriteLock;
71
import org.redisson.api.RedissonClient;
72

  
6
import redis.clients.jedis.Jedis;
7
import redis.clients.jedis.JedisPool;
73 8
import com.goldencode.p2j.util.logging.CentralLogger;
74 9

  
75
/**
76
 * A Redis-backed implementation of the {@link LockManager} interface.
77
 * <p>
78
 * This manager uses Redisson to provide distributed, hierarchical locking for 
79
 * directory tree nodes. It splits a node's path into segments and acquires 
80
 * appropriate read/write locks ("traffic" and "batch" locks) along the path 
81
 * to ensure cluster-wide consistency during concurrent operations.
82
 * </p>
83
 */
84 10
public class RedisLockManager implements LockManager 
85 11
{
86
   /** The Redisson client used to acquire distributed locks. */
87
   private RedissonClient redisson;
88
   
89
   /**
90
    * Constructs a new {@code RedisLockManager} and initializes the Redisson client
91
    * from the global {@link RedisConfig}.
92
    */
12
   private static final CentralLogger LOG = CentralLogger.get(RedisLockManager.class);
13
   
14
   private static final String JVM_ID = java.util.UUID.randomUUID().toString();
15
   private static final String LOCK_PREFIX = "dir:lock:tree:";
16
   private static final int SPIN_SLEEP_MS = 10;
17

  
18
   private JedisPool jedisPool;
19
   
20
   private static volatile String lockSha = null;
21
   private static volatile String releaseSha = null;
22
   
23
   private volatile boolean shutdown = false;
24

  
25
   private static final String LUA_LOCK_SCRIPT = 
26
   "local targetType = ARGV[1]; "                                             +
27
   "local threadId = ARGV[2]; "                                               +
28
   "for i, key in ipairs(KEYS) do "                                           +
29
       "local lockKeys = redis.call('HKEYS', key); "                          +
30
       "for j, tId in ipairs(lockKeys) do "                                   +
31
          "if tId ~= threadId then "                                          +
32
             "local lType = redis.call('HGET', key, tId); "                   +
33
             "if lType == 'WX' or targetType == 'WX' then return \"0\"; end " +
34
             "if targetType == 'RX' then return \"0\"; end "                  +
35
             "if lType == 'RX' or lType == 'INTENT_RX' then "                 +
36
                "if targetType ~= 'RO' then return \"0\"; end "               +
37
             "end "                                                           +
38
          "end "                                                              +
39
       "end "                                                                 +
40
    "end "                                                                    +
41
    "local intentType = (targetType == 'RX') and 'INTENT_RX' or 'INTENT'; "   +
42
    "for i, key in ipairs(KEYS) do "                                          +
43
       "local applyType = (i == #KEYS) and targetType or intentType; "        +
44
       "redis.call('HSET', key, threadId, applyType); "                       +
45
    "end "                                                                    +
46
    "return \"1\";";
47

  
48
   private static final String LUA_UNLOCK_SCRIPT =
49
   "local threadId = ARGV[1]; "             +
50
   "for i, key in ipairs(KEYS) do "         +
51
      "redis.call('HDEL', key, threadId); " +
52
   "end "                                   +
53
   "return \"1\";";
54

  
55
   private static final String LUA_UPGRADE_SCRIPT =
56
   "local threadId = ARGV[1]; "                                      +
57
   "local targetKey = KEYS[1]; "                                     +
58
   "local currentType = redis.call('HGET', targetKey, threadId); "   +
59
   "if currentType == 'RX' then "                                    +
60
      "if redis.call('HLEN', targetKey) > 1 then return \"0\"; end " +
61
      "redis.call('HSET', targetKey, threadId, 'WX'); "              +
62
      "return \"1\"; "                                               +
63
   "end "                                                            +
64
   "return \"0\";";
65

  
93 66
   public RedisLockManager() 
94 67
   {
95 68
   }
96 69

  
97
   /**
98
    * Acquires a distributed lock for a specific directory node and its ancestors.
99
    * <p>
100
    * The method iterates through the path hierarchy (e.g., from root down to the target node),
101
    * acquiring sequential Read/Write locks. This prevents parent nodes from being modified 
102
    * or deleted while child nodes are actively being accessed or modified.
103
    * </p>
104
    *
105
    * @param id   
106
    *        The normalized string path of the node to lock (e.g., "parent/child").
107
    * @param type 
108
    *        The type of lock requested (e.g., {@code LOCK_RW}, {@code LOCK_RX}, {@code LOCK_WX}).
109
    * 
110
    * @return An opaque lock reference object ({@link LockRef}) if the locks were successfully acquired, 
111
    *         or {@code null} if the acquisition failed or inputs were invalid.
112
    */
113 70
   @Override
114 71
   public Object enterLock(String id, int type) 
115 72
   {
116
      if (id == null || type == LOCK_WX)
117
      {
118
         return null;         
119
      }
73
      if (id == null || shutdown) return null;         
120 74
      
121 75
      id = IdUtils.normalize(id);
122
      String[] pathParts = splitPath(id);
123
      List<RLock> acquiredLocks = new ArrayList<>();
124
      
125
      if (redisson == null)
76
      List<String> keys = buildHierarchyKeys(id);
77
      String threadId = JVM_ID + ":" + Thread.currentThread().getId();
78
      String typeStr = getTypeString(type);
79
      
80
      List<String> args = new ArrayList<>();
81
      args.add(typeStr);
82
      args.add(threadId);
83
      
84
      if (jedisPool == null)
126 85
      {
127 86
         RedisConfig cfg = RedisConfig.getRedisConfig();
128
         redisson = cfg.getRedissonClient();
129
      }
130
      
131
      try 
132
      {
133
         for (int i = 0; i < pathParts.length; i++) 
134
         {
135
            String path = pathParts[i];
136
            boolean isTarget = (i == pathParts.length - 1);
137

  
138
            RReadWriteLock trafficLock = redisson.getReadWriteLock("dir:lock:traffic:" + path);
139
            RReadWriteLock batchLock = redisson.getReadWriteLock("dir:lock:batch:" + path);
140

  
141
            if (type == LOCK_RW || type == LOCK_RX) 
142
            {
143
               boolean needsTrafficWrite = isTarget && (type == LOCK_RX);
144
               RLock tLock = needsTrafficWrite ? trafficLock.writeLock() : trafficLock.readLock();
145
               
146
               tLock.lockInterruptibly();
147
               acquiredLocks.add(tLock);
148
            }
149

  
150
            RLock bLock = batchLock.readLock();
151
            bLock.lockInterruptibly();
152
            acquiredLocks.add(bLock);
153

  
154
            if (isTarget) 
155
            {
156
               return new LockRef(acquiredLocks, type, batchLock);
157
            }
158
         }
159
      }
160
      catch (InterruptedException e) 
161
      {
162
         Thread.currentThread().interrupt();
163
         releaseAll(acquiredLocks);
164
         return null;
165
      }
166
      
87
         jedisPool = cfg.getJedisPool();
88
      }
89
      
90
      while (!shutdown) 
91
      {
92
         try (Jedis jedis = jedisPool.getResource()) 
93
         {
94
            if (lockSha == null || releaseSha == null)
95
            {
96
               lockSha = jedis.scriptLoad(LUA_LOCK_SCRIPT);
97
               releaseSha = jedis.scriptLoad(LUA_UNLOCK_SCRIPT);
98
            }
99
            
100
            Object res = jedis.evalsha(lockSha, keys, args);
101
            if (res != null && "1".equals(res.toString())) 
102
            {
103
               return new LockRef(keys, type, threadId, this);
104
            }
105
         }
106
         catch (Exception e) 
107
         {
108
            LOG.warning("Transient error acquiring lock for path: " + id, e);
109
         }
110
         
111
         try 
112
         {
113
            Thread.sleep(SPIN_SLEEP_MS);
114
         } 
115
         catch (InterruptedException e) 
116
         {
117
            Thread.currentThread().interrupt();
118
            return null;
119
         }
120
      }
167 121
      return null;
168 122
   }
169 123

  
170
   /**
171
    * Releases all distributed locks associated with the provided lock reference.
172
    *
173
    * @param objLock 
174
    *        The lock reference object returned by {@link #enterLock(String, int)}.
175
    */
176 124
   @Override
177 125
   public void releaseLock(Object objLock) 
178 126
   {
......
182 130
      }
183 131
   }
184 132

  
185
   /**
186
    * Attempts to upgrade an existing read lock ({@code LOCK_RX}) to an exclusive 
187
    * write lock ({@code LOCK_WX}).
188
    *
189
    * @param objLock 
190
    *        The lock reference object to upgrade.
191
    * 
192
    * @return {@code true} if the lock was successfully upgraded, {@code false} otherwise.
193
    */
194 133
   @Override
195 134
   public boolean upgradeLock(Object objLock) 
196 135
   {
......
202 141
      return ((LockRef) objLock).upgrade();
203 142
   }
204 143

  
205
   /**
206
    * Gracefully shuts down the lock manager. 
207
    * <p>Currently, this method acts as a stub.</p>
208
    */
209 144
   @Override
210 145
   public synchronized void shutdownManager() 
211 146
   {
212
      // this.shutdown = true;
213
   }
214

  
215
   /**
216
    * Releases a list of acquired locks asynchronously in reverse order of their acquisition.
217
    *
218
    * @param locks 
219
    *        The list of Redisson locks to release.
220
    */
221
   private void releaseAll(List<RLock> locks) 
222
   {
223
      for (int i = locks.size() - 1; i >= 0; i--) 
147
      this.shutdown = true;
148
      if (jedisPool != null && !jedisPool.isClosed()) 
224 149
      {
225
         RLock lock = locks.get(i);
226
         if (lock.isHeldByCurrentThread()) 
227
         {
228
            lock.unlockAsync();
229
         }
150
         jedisPool.close();
230 151
      }
231 152
   }
232 153

  
233
   /**
234
    * Splits a hierarchical path string into an array of progressive path segments.
235
    * <p>For example, "a/b/c" becomes {@code ["a", "a/b", "a/b/c"]}.</p>
236
    *
237
    * @param id 
238
    *        The full path string to split.
239
    * 
240
    * @return An array of progressively built path strings.
241
    */
242
   private String[] splitPath(String id) 
154
   private List<String> buildHierarchyKeys(String id) 
243 155
   {
244
      if (id.isEmpty() || id.equals("/"))
156
      if (id.isEmpty() || "/".equals(id))
245 157
      {
246
         return new String[]{""};         
158
         return Collections.singletonList(LOCK_PREFIX + "");
247 159
      }
248 160
      
249 161
      String[] segments = id.split("/");
250
      String[] paths = new String[segments.length];
162
      List<String> keys = new ArrayList<>();
251 163
      StringBuilder current = new StringBuilder();
252
      for (int i = 0; i < segments.length; i++) 
164
      
165
      for (String segment : segments) 
253 166
      {
254
         if (i > 0)
167
         if (segment.isEmpty())
168
         {
169
            keys.add(LOCK_PREFIX + current.toString());
170
            continue;
171
         }
172
         
173
         if (current.length() > 0)
255 174
         {
256 175
            current.append("/");
257 176
         }
258 177
         
259
         current.append(segments[i]);
260
         paths[i] = current.toString();
178
         current.append(segment);
179
         keys.add(LOCK_PREFIX + current.toString());
261 180
      }
262 181
      
263
      return paths;
264
   }
265

  
266
   /**
267
    * An internal wrapper class that tracks the state and collection of distributed 
268
    * locks acquired during a lock operation.
269
    */
182
      return keys;
183
   }
184

  
185
   private String getTypeString(int type) 
186
   {
187
      switch (type) 
188
      {
189
         case LOCK_RX: return "RX";
190
         case LOCK_RW: return "RW";
191
         case LOCK_WX: return "WX";
192
         default:      return "RO";
193
      }
194
   }
195

  
270 196
   static class LockRef 
271 197
   {
272
      /** The sequential list of locks acquired along the node hierarchy. */
273
      private final List<RLock> locks;
274
      
275
      /** The current logical lock type (e.g., LOCK_RX, LOCK_WX). */
198
      private final List<String> keys;
276 199
      private int type;
277
      
278
      /** The specific read/write batch lock associated with the target node. */
279
      private RReadWriteLock batchLock;
200
      private final String threadId;
201
      private final RedisLockManager manager;
280 202

  
281
      /**
282
       * Creates a new lock reference.
283
       *
284
       * @param locks     
285
       *        The list of acquired locks.
286
       * @param type      
287
       *        The type of lock requested.
288
       * @param batchLock 
289
       *        The Redisson ReadWrite lock of the target node.
290
       */
291
      public LockRef(List<RLock> locks, int type, RReadWriteLock batchLock) 
203
      public LockRef(List<String> keys, int type, String threadId, RedisLockManager manager) 
292 204
      {
293
         this.locks = locks;
205
         this.keys = keys;
294 206
         this.type = type;
295
         this.batchLock = batchLock;
207
         this.threadId = threadId;
208
         this.manager = manager;
296 209
      }
297 210

  
298
      /**
299
       * Upgrades a {@code LOCK_RX} (Read Exclusive) lock to a {@code LOCK_WX} (Write Exclusive) lock.
300
       * It releases the current read lock and blocks until the write lock can be acquired.
301
       *
302
       * @return {@code true} if upgraded successfully, {@code false} if the current lock 
303
       *         type does not support upgrading.
304
       */
305 211
      public boolean upgrade() 
306 212
      {
307
         if (type == LOCK_RX)
213
         if (this.type != LOCK_RX) return false;
214

  
215
         List<String> targetKeyList = Collections.singletonList(keys.get(keys.size() - 1));
216
         List<String> args = new ArrayList<>();
217
         args.add(threadId);
218
         
219
         while (!manager.shutdown) 
308 220
         {
221
            try (Jedis jedis = manager.jedisPool.getResource())
222
            {
223
               Object res = jedis.eval(LUA_UPGRADE_SCRIPT, targetKeyList, args);
224
               if (res != null && "1".equals(res.toString())) 
225
               {
226
                  this.type = LOCK_WX;
227
                  return true;
228
               }
229
            }
230
            catch (Exception e) 
231
            {
232
               LOG.warning("Exception during lock upgrade attempt", e);
233
            }
234
            
309 235
            try 
310 236
            {
311
               RLock rdLock = batchLock.readLock();
312
               RLock wxLock = batchLock.writeLock();
313
               
314
               rdLock.unlock();
315
               wxLock.lockInterruptibly(); 
316
               
317
               locks.add(wxLock);
318
               this.type = LOCK_WX;
319
               return true;
320
            }
237
               Thread.sleep(SPIN_SLEEP_MS);
238
            } 
321 239
            catch (InterruptedException e) 
322 240
            {
323 241
               Thread.currentThread().interrupt();
242
               return false;
324 243
            }
325 244
         }
326
         
327 245
         return false;
328 246
      }
329 247

  
330
      /**
331
       * Synchronously releases all locks tracked by this reference in reverse order 
332
       * of their acquisition, ensuring safe tear-down of hierarchical locks.
333
       */
334 248
      public void release() 
335 249
      {
336
         for (int i = locks.size() - 1; i >= 0; i--) 
337
         {
338
            RLock lock = locks.get(i);
339
            if (lock.isHeldByCurrentThread()) 
340
            {
341
               lock.unlock();
342
            }
250
         try (Jedis jedis = manager.jedisPool.getResource())
251
         {
252
            List<String> args = Collections.singletonList(threadId);
253
            jedis.evalsha(releaseSha, keys, args);
254
         } 
255
         catch (Exception e) 
256
         {
257
            LOG.warning("Failed to clean up keys in Redis on release", e);
343 258
         }
344 259
      }
345 260
   }
new/src/com/goldencode/p2j/directory/RedisRemapper.java 2026-05-18 06:10:11 +0000
101 101
   private static final String KEY_ATTR = "dir:attr:%s:%s";
102 102
   
103 103
   /** Global Redis key used for synchronization during initialization/load. */
104
   private static final String LOCK_KEY = "dir:lock:batch:";
104
   private static final String LOCK_KEY = "dir:lock:tree:";
105 105
   
106 106
   /** Redis key used to flag that the system is under maintenance (read-only mode). */
107 107
   private static final String MAINTENANCE = "maintenance";
......
112 112
   /** XML attribute name used to store the structural version/revision in the root element. */
113 113
   private static final String VERSION_ATTR_NAME = "version";
114 114
   
115
   /**
116
    * Atomically checks if a load is in progress. If not, it wipes the DB,
117
    * sets the lock, and returns 1. If someone else is loading, it returns 0.
118
    */
119
   private static final String LUA_FORCE_LOAD_INIT =
120
      "local lockKey = KEYS[1]; "                   +
121
      "local threadId = ARGV[1]; "                  +
122
      "redis.call('FLUSHDB'); "                     +
123
      "redis.call('HSET', lockKey, threadId, 'WX')" +
124
      "return \"1\";";
125
   
126
   private static final String LUA_RELEASE_LOAD_LOCK =
127
   "local threadId = ARGV[1]; " +
128
   "for i, key in ipairs(KEYS) do " +
129
   "redis.call('HDEL', key, threadId); " +
130
   "end " +
131
   "return \"1\";";
132
   
115 133
   /** JSON Mapper used to serialize/deserialize {@link Attribute} objects. */
116 134
   private static final ObjectMapper json_mapper = new ObjectMapper();
117 135

  
......
272 290
   throws Exception
273 291
   {
274 292
      if (!initialize)
275
      {      
276
         RReadWriteLock lock = redissonClient.getReadWriteLock(LOCK_KEY);
293
      {
294
         String threadId = java.util.UUID.randomUUID().toString();
277 295
         
278
         try
296
         try (Jedis jedis = jedisPool.getResource())
279 297
         {
280
            lock.writeLock().lockInterruptibly();
298
            Object res = jedis.eval(LUA_FORCE_LOAD_INIT, 
299
                                    Collections.singletonList(LOCK_KEY), 
300
                                    Collections.singletonList(threadId));
281 301
            
282
            try (Jedis jedis = jedisPool.getResource())
283
            {
284
               if (file == null || !file.exists() || !file.isFile())
302
            try
303
            {               
304
               if (res != null && "1".equals(res.toString()))
285 305
               {
286
                  return;
306
                  switch (res.toString())
307
                  {
308
                     case "1":
309
                     {
310
                        if (file == null || !file.exists() || !file.isFile())
311
                        {
312
                           return;
313
                        }
314
                        
315
                        XmlRedisRemapperIO io = new XmlRedisRemapperIO(this, jedis);
316
                        io.load(new FileInputStream(file));
317
                        initialize = true;                     
318
                     }
319
                  }
287 320
               }
288
               
289
               XmlRedisRemapperIO io = new XmlRedisRemapperIO(this, jedis);
290
               io.load(new FileInputStream(file));
291
               initialize = true;
292 321
            }
293
         }
294
         finally
295
         {
296
            if (lock != null && lock.writeLock().isHeldByCurrentThread())
322
            finally
297 323
            {
298
               lock.writeLock().unlock();
324
               jedis.eval(LUA_RELEASE_LOAD_LOCK, 
325
                        Collections.singletonList(LOCK_KEY), 
326
                        Collections.singletonList(threadId));
299 327
            }
300 328
         }
301 329
      }