EventSemaphore.java
/*
** Module : EventSemaphore.java
** Abstract : event semaphore concurrency primitive
**
** Copyright (c) 2020, Golden Code Development Corporation.
**
** -#- -I- --Date-- ---------------------------------Description---------------------------------
** 001 GES 20200116 First 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.concurrent;
import java.util.*;
import java.util.function.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.util.*;
/**
* Implements an event semaphore concurrency primitive.
* <p>
* The idea here is to provide the capability of posting (or raising) a signal which can be
* waited on by other threads. An event semaphore has two possible states: reset (the event
* has not been signaled) and posted (the event has occurred). Threads call {@code await} to
* wait for the event to be posted.
* <p>
* {@code reset} is used to move the semaphore into reset state and {@code post} is used to
* move the semaphore into posted state.
* <p>
* {@code await} will return immediately if the semaphore is already posted, but it will block
* if the semaphore is in reset state. All waiting threads are released at the moment that
* the semaphore shifts from reset into posted state. If the semaphore is reset after posting
* and a thread that was waiting before the post occurred does not run before the reset happens,
* that thread will still be released and it will still be runnable. In other words, this
* mechanism is "edge triggered" and is NOT "level" based.
* <p>
* The {@code await} method has a form that waits indefinitely and a second form that takes a
* timeout value in milliseconds. Upon return from {@code await} a status enum is returned.
* There are 4 ways to exit await (posted, timeout, interrupted, semaphore destroyed).
* <p>
* If you need to check to see if the {@code await} will block, you may call {@code query} to
* read the posted count. <b>WARNING: there is no guarantee that the state will remain the same
* after the call to {@code query} returns.</b>
*/
public final class EventSemaphore
extends RelatedResource
{
/** Unique resource type name, used as a namespace prefix. */
private static final String RESOURCE_TYPE = "event_semaphore";
/** Number of times this semaphore has been posted. 0 means reset. */
private int posted = 0;
/**
* Disable external construction.
*
* @param name
* Resource name, must not be {@code null}.
*/
private EventSemaphore(String name)
{
super(name);
}
/**
* Factory method which creates a new local event semaphore with the given name. The
* semaphore will be in the reset state at creation.
*
* @param name
* The semaphore name, without the standard "/event_semaphore/" prefix.
*
* @return The new semaphore if it was created.
*
* @throws ErrorConditionException
* If there was any problem in creation.
*/
public static synchronized EventSemaphore create(String name)
throws ErrorConditionException
{
return create(name, true);
}
/**
* Factory method which creates a new local event semaphore with the given name.
*
* @param name
* The semaphore name, without the standard "/event_semaphore/" prefix.
* @param reset
* {@code true} to set the state to reset at creation, {@code false} to set the
* initial state to posted.
*
* @return The new semaphore if it was created.
*
* @throws ErrorConditionException
* If there was any problem in creation.
*/
public static synchronized EventSemaphore create(String name, boolean reset)
throws ErrorConditionException
{
return create(name, reset, true);
}
/**
* Factory method which creates a new event semaphore with the given name.
*
* @param name
* The semaphore name, without the standard "/event_semaphore/" prefix.
* @param reset
* {@code true} to set the state to reset at creation, {@code false} to set the
* initial state to posted.
* @param local
* {@code true} to create a local instance (one that is only accessible between
* threads in the same security context. {@code false} creates an instance that
* is accessible from any session in the JVM.
*
* @return The new semaphore if it was created.
*
* @throws ErrorConditionException
* If there was any problem in creation.
*/
public static synchronized EventSemaphore create(String name, boolean reset, boolean local)
throws ErrorConditionException
{
Function<String, ? extends RelatedResource> creator = (String clean) ->
{
EventSemaphore event = new EventSemaphore(clean);
if (!reset)
{
event.posted = 1;
}
return event;
};
return (EventSemaphore) RelatedResource.create(RESOURCE_TYPE, name, creator, local);
}
/**
* Opens an existing local event semaphore with the given name.
*
* @param name
* The semaphore name, without the standard "/event_semaphore/" prefix.
*
* @return The semaphore if it exists and was opened.
*
* @throws ErrorConditionException
* If there was any problem in opening.
*/
public static synchronized EventSemaphore open(String name)
throws ErrorConditionException
{
return open(name, true);
}
/**
* Opens an existing local event semaphore with the given name.
*
* @param name
* The semaphore name, without the standard "/event_semaphore/" prefix.
* @param local
* {@code true} to open a local instance (one that is only accessible between
* threads in the same security context. {@code false} opens an instance that
* is accessible from any session in the JVM.
*
* @return The semaphore if it exists and was opened.
*
* @throws ErrorConditionException
* If there was any problem in opening.
*/
public static synchronized EventSemaphore open(String name, boolean local)
throws ErrorConditionException
{
return (EventSemaphore) RelatedResource.open(RESOURCE_TYPE, name, local);
}
/**
* Obtain the resource name (which makes up the prefix for valid names of this resource type).
*
* @return The non-null, non-empty resource type without spaces or '/' characters. This
* must be the same for all resources of this type.
*/
public String resourceName()
{
return RESOURCE_TYPE;
}
/**
* Increase the posted count by 1 and if the old posted count was 0, release all waiting
* threads. Please note that the threads are only released when the post count changes
* from 0 to 1. This means that if the semaphore is reset after posting and a thread that
* was waiting before the post occurred does not run before the reset happens, the thread
* will still be released and it will still be runnable.
*
* @throws ErrorConditionException
* If the semaphore is destroyed or if there is an access rights issue.
*/
public void post()
throws ErrorConditionException
{
errorIfDead();
checkWrite();
synchronized (lock)
{
posted++;
// we are edge triggered
if (posted == 1)
{
release();
}
}
}
/**
* Block until the semaphore is posted (if the semaphore is current reset) and if the
* semaphore is already posted at the time this is called, then return immediately.
* <p>
* This version of the waiting will not time out.
*
* @throws ErrorConditionException
* If the semaphore is destroyed or if there is an access rights issue.
*/
public WaitStatus await()
throws ErrorConditionException
{
return await(-1);
}
/**
* Block until the semaphore is posted (if the semaphore is current reset) OR until the
* specified number of milliseconds has elapsed, whichever comes first. If the
* semaphore is already posted at the time this is called, then return immediately.
*
* @param millis
* Maximum wait time, use 0 or any negative number for an indefinite wait.
*
* @throws ErrorConditionException
* If the semaphore is destroyed or if there is an access rights issue.
*/
public WaitStatus await(long millis)
throws ErrorConditionException
{
errorIfDead();
checkWrite();
if (millis < 0)
{
millis = 0;
}
WaitStatus status = WaitStatus.OK;
synchronized (lock)
{
long start = System.currentTimeMillis();
while (active && posted == 0)
{
try
{
lock.wait(millis);
long elapsed = System.currentTimeMillis() - start;
if (millis > 0 && elapsed >= millis)
{
status = WaitStatus.TIMEOUT;
break;
}
}
catch (InterruptedException ie)
{
status = WaitStatus.INTERRUPTED;
break;
}
}
if (!active)
{
status = WaitStatus.DESTROYED;
}
}
return status;
}
/**
* Reset the posted count and return the old count value.
*
* @return The previous posted count. The semaphore was already in reset state if this
* value is 0.
*
* @throws ErrorConditionException
* If the semaphore is destroyed or if there is an access rights issue.
*/
public int reset()
throws ErrorConditionException
{
errorIfDead();
checkWrite();
int old = 0;
synchronized (lock)
{
old = posted;
posted = 0;
}
return old;
}
/**
* Return the number of times the semaphore has been posted since the last reset.
* <p>
* This can be used to detect if a call to {@code await} would block AT THIS INSTANT.
* <b>WARNING: there is no guarantee that the state will remain the same after the
* call to returns, so an immediate call to {@code await} will block if the semaphore
* is {@code reset} in between the {@code query} and {@code await} calls.</b>
*
* @return The posted count. The semaphore is reset if this value is 0.
*
* @throws ErrorConditionException
* If the semaphore is destroyed or if there is an access rights issue.
*/
public int query()
throws ErrorConditionException
{
errorIfDead();
checkRead();
synchronized (lock)
{
return posted;
}
}
}