Merge "Updated function test environment"
[nonrtric.git] / policy-agent / src / main / java / org / oransc / policyagent / repository / Lock.java
1 /*-
2  * ========================LICENSE_START=================================
3  * O-RAN-SC
4  * %%
5  * Copyright (C) 2019 Nordix Foundation
6  * %%
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  *
11  *      http://www.apache.org/licenses/LICENSE-2.0
12  *
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  * ========================LICENSE_END===================================
19  */
20
21 package org.oransc.policyagent.repository;
22
23 import java.util.ArrayList;
24 import java.util.Iterator;
25 import java.util.LinkedList;
26 import java.util.List;
27
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30
31 import reactor.core.publisher.Mono;
32 import reactor.core.publisher.MonoSink;
33
34 /**
35  * A resource lock. Exclusive means that the caller takes exclusive ownership of
36  * the resurce. Non exclusive lock means that several users can lock the
37  * resource (for shared usage).
38  */
39 public class Lock {
40     private static final Logger logger = LoggerFactory.getLogger(Lock.class);
41
42     private boolean isExclusive = false;
43     private int lockCounter = 0;
44     private final List<LockRequest> lockRequestQueue = new LinkedList<>();
45     private static AsynchCallbackExecutor callbackProcessor = new AsynchCallbackExecutor();
46
47     private static class AsynchCallbackExecutor implements Runnable {
48         private List<LockRequest> lockRequestQueue = new LinkedList<>();
49
50         public AsynchCallbackExecutor() {
51             Thread thread = new Thread(this);
52             thread.start();
53         }
54
55         public synchronized void addAll(List<LockRequest> requests) {
56             this.lockRequestQueue.addAll(requests);
57             this.notifyAll();
58         }
59
60         @Override
61         public void run() {
62             try {
63                 while (true) {
64                     for (LockRequest request : consume()) {
65                         request.callback.success(request.lock);
66                     }
67                     waitForNewEntries();
68                 }
69             } catch (InterruptedException e) {
70                 Thread.currentThread().interrupt();
71                 logger.error("Interrupted {}", e.getMessage());
72             }
73         }
74
75         private synchronized List<LockRequest> consume() {
76             List<LockRequest> q = this.lockRequestQueue;
77             this.lockRequestQueue = new LinkedList<>();
78             return q;
79         }
80
81         @SuppressWarnings("java:S2274")
82         private synchronized void waitForNewEntries() throws InterruptedException {
83             if (this.lockRequestQueue.isEmpty()) {
84                 this.wait();
85             }
86         }
87     }
88
89     public enum LockType {
90         EXCLUSIVE, SHARED
91     }
92
93     /** The caller thread will be blocked util the lock is granted. */
94     public synchronized void lockBlocking(LockType locktype) {
95         while (!tryLock(locktype)) {
96             this.waitForUnlock();
97         }
98     }
99
100     /** Reactive version. The Lock will be emitted when the lock is granted */
101     public synchronized Mono<Lock> lock(LockType lockType) {
102         if (tryLock(lockType)) {
103             return Mono.just(this);
104         } else {
105             return Mono.create(monoSink -> addToQueue(monoSink, lockType));
106         }
107     }
108
109     public Mono<Lock> unlock() {
110         return Mono.create(monoSink -> {
111             unlockBlocking();
112             monoSink.success(this);
113         });
114     }
115
116     public void unlockBlocking() {
117         synchronized (this) {
118             if (lockCounter <= 0) {
119                 lockCounter = -1; // Might as well stop, to make it easier to find the problem
120                 logger.error("Number of unlocks must match the number of locks");
121             }
122             this.lockCounter--;
123             if (lockCounter == 0) {
124                 isExclusive = false;
125             }
126             this.notifyAll();
127         }
128         this.processQueuedEntries();
129     }
130
131     @Override
132     public String toString() {
133         return "Lock cnt: " + this.lockCounter + " exclusive: " + this.isExclusive + " queued: "
134             + this.lockRequestQueue.size();
135     }
136
137     /** returns the current number of granted locks */
138     public synchronized int getLockCounter() {
139         return this.lockCounter;
140     }
141
142     private void processQueuedEntries() {
143         List<LockRequest> granted = new ArrayList<>();
144         synchronized (this) {
145             for (Iterator<LockRequest> i = lockRequestQueue.iterator(); i.hasNext();) {
146                 LockRequest request = i.next();
147                 if (tryLock(request.lockType)) {
148                     i.remove();
149                     granted.add(request);
150                 }
151             }
152         }
153         callbackProcessor.addAll(granted);
154     }
155
156     private static class LockRequest {
157         final MonoSink<Lock> callback;
158         final LockType lockType;
159         final Lock lock;
160
161         LockRequest(MonoSink<Lock> callback, LockType lockType, Lock lock) {
162             this.callback = callback;
163             this.lockType = lockType;
164             this.lock = lock;
165         }
166     }
167
168     private synchronized void addToQueue(MonoSink<Lock> callback, LockType lockType) {
169         lockRequestQueue.add(new LockRequest(callback, lockType, this));
170     }
171
172     @SuppressWarnings("java:S2274") // Always invoke wait() and await() methods inside a loop
173     private synchronized void waitForUnlock() {
174         try {
175             this.wait();
176         } catch (InterruptedException e) {
177             logger.warn("waitForUnlock interrupted", e);
178             Thread.currentThread().interrupt();
179         }
180     }
181
182     private boolean tryLock(LockType lockType) {
183         if (this.isExclusive) {
184             return false;
185         }
186         if (lockType == LockType.EXCLUSIVE && lockCounter > 0) {
187             return false;
188         }
189         lockCounter++;
190         this.isExclusive = lockType == LockType.EXCLUSIVE;
191         return true;
192     }
193
194 }