浏览代码

Change HALT to PAUSE

Added syncrhonized qualifier to 2 methods to guard against simultaneous update by 2 threads.
Completed CANCEL processing.
tags/Production_2014_08_01
ymlam 11 年前
父节点
当前提交
83b9ecefad
共有 1 个文件被更改,包括 98 次插入53 次删除
  1. +98
    -53
      src/altk/comm/engine/Broadcast.java

+ 98
- 53
src/altk/comm/engine/Broadcast.java 查看文件

@@ -1,7 +1,6 @@
package altk.comm.engine;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -18,7 +17,6 @@ import javax.servlet.http.HttpServletRequest;

import org.apache.log4j.Logger;
import org.apache.log4j.NDC;
import org.omg.CORBA_2_3.portable.OutputStream;

import altk.comm.engine.exception.BroadcastException;
import altk.comm.engine.exception.EngineException;
@@ -38,7 +36,7 @@ public abstract class Broadcast
public final String broadcastType;
private String broadcastId;
private BroadcastState state = BroadcastState.INSTALLING;
private BroadcastState state = BroadcastState.ACCEPTED;

String haltReason;
String stateErrorText;
@@ -60,9 +58,6 @@ public abstract class Broadcast

protected static Logger myLogger = Logger.getLogger(Broadcast.class);

/**
* This queue is designed for only one server
*/
private Queue<Job> readyQueue;
private List<Thread> serviceThreadPool;
private Object resumeFlag; // Semaphore for dispatcher threads to resume.
@@ -74,15 +69,28 @@ public abstract class Broadcast

public static enum BroadcastState
{
INSTALLING,
ACCEPTED,
RUNNING,
HALTING,
HALTED,
PAUSING,
PAUSED,
CANCELING,
CANCELED, // Final state
PURGED, // Final state
ABORTED, // final state
COMPLETED // Final state
CANCELED(true), // Final state
PURGED(true), // Final state
ABORTED(true), // final state
EXPIRED(true), // final state
COMPLETED(true); // Final state
final public boolean isFinal;
private BroadcastState()
{
isFinal = false;
}
private BroadcastState(boolean isFinal)
{
this.isFinal = isFinal;
}
}
public enum StateChangeStatus
@@ -144,12 +152,12 @@ public abstract class Broadcast
toStates = new HashMap<BroadcastState, List<BroadcastState>>();

// Transitions from INSTALLING
toStates.put(BroadcastState.INSTALLING, Arrays.asList(
toStates.put(BroadcastState.ACCEPTED, Arrays.asList(
BroadcastState.RUNNING, // Normal transition
BroadcastState.CANCELING, // User action
BroadcastState.CANCELED, // User action
BroadcastState.HALTING, // User action
BroadcastState.HALTED, // User action
BroadcastState.PAUSING, // User action
BroadcastState.PAUSED, // User action
BroadcastState.PURGED, // User action
BroadcastState.ABORTED, // TTS error
BroadcastState.COMPLETED // When recipient list is empty
@@ -159,8 +167,8 @@ public abstract class Broadcast
toStates.put(BroadcastState.RUNNING, Arrays.asList(
BroadcastState.CANCELING, // User action
BroadcastState.CANCELED, // User action
BroadcastState.HALTING, // User action
BroadcastState.HALTED, // User action
BroadcastState.PAUSING, // User action
BroadcastState.PAUSED, // User action
BroadcastState.PURGED, // User action
BroadcastState.ABORTED, // Service provider irrecoverable error
BroadcastState.COMPLETED // Natural transition, if all ongoing calls complete and no more calls in Dispatcher queues.
@@ -174,16 +182,16 @@ public abstract class Broadcast
));
// Transitions from HALTING
toStates.put(BroadcastState.HALTING, Arrays.asList(
toStates.put(BroadcastState.PAUSING, Arrays.asList(
BroadcastState.RUNNING, // User action
BroadcastState.CANCELED, // User action
BroadcastState.HALTED,
BroadcastState.PAUSED,
BroadcastState.PURGED, // User action
BroadcastState.COMPLETED // Natural transition, if all ongoing jobs complete and no more calls in Dispatcher queues.
));

// Transitions from HALTED
toStates.put(BroadcastState.HALTED, Arrays.asList(
toStates.put(BroadcastState.PAUSED, Arrays.asList(
BroadcastState.RUNNING, // User action
BroadcastState.CANCELED, // User action
BroadcastState.CANCELING, // User action
@@ -220,8 +228,7 @@ public abstract class Broadcast
public void run()
{
myLogger.info("Thread " + getName() + " starting...");
NDC.push(getName());
myLogger.info("Thread starting...");
for (;;)
{
if (threadsShouldStop())
@@ -269,6 +276,16 @@ public abstract class Broadcast
return;
}
// Check if expired
if (System.currentTimeMillis() >= expireTime)
{
setState(BroadcastState.EXPIRED);
continue;
}
/**
* Includes allocation from capacity. Only returns when the required allocation
* is obtained. Example, RTP port allocation, limit due to total number of allowable calls.
*/
prerequisites = secureServicePrerequisites();
if (threadsShouldStop() || threadsShouldPause())
@@ -277,6 +294,15 @@ public abstract class Broadcast
continue;
}
// Check again if expired
if (System.currentTimeMillis() >= expireTime)
{
returnPrerequisites(prerequisites);
setState(BroadcastState.EXPIRED);
continue;
}

// Now that we can go ahead with this job, let us remove this from queue
readyQueue.poll();
@@ -304,7 +330,6 @@ public abstract class Broadcast
try
{
processJobs(batch, serviceProviderPeer, prerequisites);
completedJobCount++;
}
catch (EngineException e)
{
@@ -374,11 +399,11 @@ public abstract class Broadcast
/**
* Makes a state transition to the given newState if the transition from
* the current state is legal.
* the current state is legal. Also posts back a state change notification.
* @param newState
* @return StateChangeResult
*/
public StateChangeResult setState(BroadcastState newState,
public synchronized StateChangeResult setState(BroadcastState newState,
String haltReason, String stateErrorText)
{
boolean isLegal;
@@ -529,13 +554,7 @@ public abstract class Broadcast
return responseXML.toString();
}

private boolean stateIsFinal(BroadcastState state)
{
return state == BroadcastState.ABORTED || state == BroadcastState.CANCELED
|| state == BroadcastState.COMPLETED || state == BroadcastState.PURGED;
}
/**
/**
* If finalState is final, then this state is set, and dispatcher threads are stopped.
* Overriding implementation may release all other resources, like timers.
* @param finalState
@@ -552,8 +571,20 @@ public abstract class Broadcast
*/
public void terminate(BroadcastState finalState, String reason)
{
if (!stateIsFinal(finalState)) throw new IllegalArgumentException("Argument finalState " + finalState + " in Broadcast.terminate method is not final");
setState(finalState, reason, null);
if (!finalState.isFinal) throw new IllegalArgumentException("Argument finalState " + finalState + " in Broadcast.terminate method is not final");
StateChangeResult result = setState(finalState, reason, null);
switch (result.stateChangeStatus)
{
case SUCCESS:
break;
case NO_CHANGE:
return;
case FORBIDDEN:
myLogger.error("Not allow to terminate broadcast in " + result.currentState + " state");
return;
default: // Should not happen
return;
}
// Wake up all dispatcher threads waiting on readyQueue so they will all stop
synchronized(readyQueue)
@@ -597,7 +628,7 @@ public abstract class Broadcast
}
statusBf.append(">\r\n<state>" + state + "</state><state_change_time>" + changeStateTime
+ "</state_change_time>\r\n");
if (state == BroadcastState.HALTED
if (state == BroadcastState.PAUSED
|| state == BroadcastState.ABORTED)
{
if (haltReason != null)
@@ -706,11 +737,25 @@ public abstract class Broadcast
/**
* Sets the stateMachine to CANCEL
*/
protected void cancel()
protected void cancel(PrintWriter out)
{
if (this.getActiveJobCount() == 0) setState(BroadcastState.CANCELED);
// Sets state to CANCELING, which is monitored by its Broadcast.Service threads.
else setState(BroadcastState.CANCELING);
BroadcastState targetState = getActiveJobCount() == 0?
BroadcastState.CANCELED : BroadcastState.CANCELING;
StateChangeResult result = setState(targetState);
String responseContent = null;
switch (result.stateChangeStatus)
{
case SUCCESS:
responseContent = "OK";
break;
case NO_CHANGE:
responseContent = "Not canceled: Already cancelled";
break;
case FORBIDDEN:
responseContent = "Not canceled: Not allowed to cancel a broadcast in " + result.currentState + " state";
}
out.write(responseContent);

synchronized(resumeFlag)
{
resumeFlag.notifyAll();
@@ -720,7 +765,7 @@ public abstract class Broadcast
protected void pause()
{
// Sets state to HALTED, which is monitored by Broadcast.Service threads.
setState(BroadcastState.HALTING);
setState(BroadcastState.PAUSING);
}
protected void resume()
@@ -775,13 +820,13 @@ public abstract class Broadcast
private boolean threadsShouldStop()
{
BroadcastState state = getState();
return state == BroadcastState.CANCELING || stateIsFinal(state);
return state == BroadcastState.CANCELING || state.isFinal;
}

private boolean threadsShouldPause()
{
BroadcastState state = getState();
return state == BroadcastState.HALTED || state == BroadcastState.HALTING;
return state == BroadcastState.PAUSED || state == BroadcastState.PAUSING;
}
/**
@@ -804,12 +849,13 @@ public abstract class Broadcast

/**
* Sets jobStatus in job, and post job report.
* If no rescheduling, then decrement number of remainingJobs,
* If jobStatus is final, and no rescheduling,
* then decrement number of remainingJobs,and increment completedJobCount.
* @param job
* @param jobStatus
* @param errorText
*/
public void postJobStatus(Job job)
public synchronized void postJobStatus(Job job)
{
if (postBack != null)
{
@@ -822,6 +868,8 @@ public abstract class Broadcast
if (job.jobStatus.isTerminal())
{
remainingJobs--;
completedJobCount++;

if (remainingJobs == 0)
{
terminate(BroadcastState.COMPLETED);
@@ -829,13 +877,14 @@ public abstract class Broadcast
else if (getActiveJobCount() == 0)
{
if (state == BroadcastState.CANCELING) setState(BroadcastState.CANCELED);
else if (state == BroadcastState.HALTING) setState(BroadcastState.HALTED);
else if (state == BroadcastState.PAUSING) setState(BroadcastState.PAUSED);
}
}
}
/**
* Sets jobStatus in job, and post job report.
* Optionally reschedules job.
* If no rescheduling, then decrement number of remainingJobs,
* @param job
* @param jobStatus
@@ -845,10 +894,6 @@ public abstract class Broadcast
protected void postJobStatus(Job job, long rescheduleTimeMS)
{
postJobStatus(job);
if (rescheduleTimeMS < 0)
{
completedJobCount++;
}
if (rescheduleTimeMS == 0)
{
addJob(job);
@@ -875,8 +920,8 @@ public abstract class Broadcast
switch (state)
{
case RUNNING:
case HALTING:
case HALTED:
case PAUSING:
case PAUSED:
return readyQueue.size();
default:
return 0;


正在加载...
取消
保存