asterisk-java-users

[Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Johan S. <js...@sv...>]

Hi dear users and experts J

What about my problem, any clever ideas?
I've got 10 phonelines with incoming calls, and DTMF-tones are coming too on these lines.

I've written an answering AGI-javaprogram, to service() each call in parallel.
Example of what happens during one call:
Call 1 - answered by my java-class(via AGI-call),
I'm sending some DTMF-tones to the caller (via my AsteriskChannel and obj.playDtmf)and I will shortly then receive a sequence of DTMF-tones as the answer from the caller. (using agiChannel.waitForDigit(waitDelay))

What happens is, when many channels are in progress simultaneously sending DTMF, and listening for DTMF-responses, each thread will not run often enough to get all incoming DTMF-tones.
It could take like 10 seconds (looking at my log) before one thread sent the DTMF-question, till it calls the get-DTMF()-command, but then it's too late.
The delay seems to be all other threads executing in between.
It's the same situation for every Thread.

I wonder, why is not every thread being switched more often? Like once every millisecond?
It's more like once every second here.
Surely something must be wrong... I must be a very bad programmer... J

With just one CPU, multithreaded tasks need to switch thread very quick, to work.

What have I done wrong? Help, opinions are appreciated.

Sincerely,
Johan

__________________________________________________________________
Johan Sandgren
Svep Design Center AB (www.svep.se<http://www.svep.se>)
St. Lars väg 42A, SE-222 70 Lund
Phone: 046-19 27 22

Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Gopal k. <gop...@pe...>]

Hi,

  we have done the dtmf application with asterisk-java using "getdata()" let
me check whether it will support parallel inbound calls.

On Thu, Feb 14, 2008 at 6:16 PM, Johan Sandgren <js...@sv...> wrote:

>  Hi dear users and experts J
>
>
>
> What about my problem, any clever ideas?
>
> I've got 10 phonelines with incoming calls, and DTMF-tones are coming too
> on these lines.
>
>
>
> I've written an answering AGI-javaprogram, to service() each call in
> parallel.
>
> *Example of what happens during one call:*
> Call 1 – answered by my java-class(via AGI-call),
> I'm sending some DTMF-tones to the caller (via my AsteriskChannel and
> obj.playDtmf)and I will shortly then receive a sequence of DTMF-tones as
> the answer from the caller. (using *agiChannel.waitForDigit(waitDelay))*
>
>
>
> What happens is, when many channels are in progress simultaneously sending
> DTMF, and listening for DTMF-responses, each thread will not run often
> enough to get all incoming DTMF-tones.
>
> It could take like 10 seconds (looking at my log) before one thread sent
> the DTMF-question, till it calls the get-DTMF()-command, but then it's too
> late.
>
> The delay seems to be all other threads executing in between.
>
> It's the same situation for every Thread.
>
>
>
> I wonder, why is not every thread being switched more often? Like once
> every millisecond?
>
> It's more like once every second here.
>
> Surely something must be wrong… I must be a very bad programmer… J
>
>
>
> With just one CPU, multithreaded tasks need to switch thread very quick,
> to work.
>
>
>
> What have I done wrong? Help, opinions are appreciated.
>
>
>
> Sincerely,
>
> Johan
>
>
>
> __________________________________________________________________
>
>
> Johan Sandgren
>
> Svep Design Center AB (www.svep.se)
>
> St. Lars väg 42A, SE-222 70 Lund
>
> Phone: 046-19 27 22
>
>
>
> -------------------------------------------------------------------------
> This SF.net email is sponsored by: Microsoft
> Defy all challenges. Microsoft(R) Visual Studio 2008.
> http://clk.atdmt.com/MRT/go/vse0120000070mrt/direct/01/
> _______________________________________________
> Asterisk-java-users mailing list
> Ast...@li...
> https://lists.sourceforge.net/lists/listinfo/asterisk-java-users
>
>


-- 
Thank you with regards,
Gopal,
PeopleTech Systems Private Limited
www.peopletech.co.in

Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Stefan R. <ste...@re...>]

Johan Sandgren wrote:
> What about my problem, any clever ideas?

would you mind to post your code?

=Stefan

-- 
reuter network consulting
Neusser Str. 110
50760 Koeln
Germany
Telefon: +49 221 1305699-0
Telefax: +49 221 1305699-90
E-Mail:  ste...@re...
Jabber:  ste...@re...
WWW:     http://www.reucon.com

Steuernummern 215/5140/1791 USt-IdNr. DE220701760

Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Julien T. <om...@gm...>]

Stefan Reuter wrote:
> would you mind to post your code?

Hi,

I'm sure that thread will interest me as well.
Please, post your code so that Stefan can provide some feedback.

Actually, I never posted any question on the ML but I am not so sure of 
me on how Asterisk-Java handles threads. Think it's one instance of 
AgiScript for each concurrent thread? So each instance variable access 
(write?) has to be synchronized?

My main example would be something like...

public class PlopScript extends BaseAgiScript {

     private Connection db;

     // constructor
     public PlopScript() {

         // initialize database connection object
         db = DriverManager.getConnection(...);
     }

     public void service(AgiRequest request, AgiChannel channel)
     throws AgiException {

         // ....
         //
         // so now if i want to query my DB, i need to synchronize
         // each call using my "db" object? i am correct?
         //
         // i mean... let's assume I will have concurrent calls
         // executing that PlopScript instance
     }

}

I know you would say, "it's better/easier to declare variables/objects 
IN the service method". But, just to be sure... If I would like to have 
instance members...

Let's see Johan's examples and then Stefan (or other gurus) answer, 
think that will help me...

Cheers,
Julien

Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Stefan R. <ste...@re...>]

Julien,

thanks for your question.

> Actually, I never posted any question on the ML but I am not so sure of 
> me on how Asterisk-Java handles threads. Think it's one instance of 
> AgiScript for each concurrent thread? So each instance variable access 
> (write?) has to be synchronized?

Asterisk-Java actually supports both models:
a) one shared instance for all calls
b) one instance per call

The preferred model is (a). It is also the default.
In your example that would mean there is only one db connection that is
shared by all calls. If you replace the db connection with a connection
pool that is initialized in the contructor, that would match quite well.
For instance fields the usual thread safety rules apply:
- syncronized write access (if any, usually the instance members are
final and initialized in the ctor)
- local variables in the service() method for call specific things.
Of cource your service method could also create new objects and delegate
to them.

Model (b) creates a new instance for each call so you don't have any
shared state at the cost of "bootstrapping" you script for each call.

To switch to model (b) you have to set the shareInstances property of
your mapping strategy to false. See
http://asterisk-java.org/development/apidocs/org/asteriskjava/fastagi/ResourceBundleMappingStrategy.html
http://asterisk-java.org/development/apidocs/org/asteriskjava/fastagi/ClassNameMappingStrategy.html

Independant of whether you use shared instances or not each call is
always executed in its own thread.

I hope this makes things a bit clearer.

=Stefan

-- 
reuter network consulting
Neusser Str. 110
50760 Koeln
Germany
Telefon: +49 221 1305699-0
Telefax: +49 221 1305699-90
E-Mail:  ste...@re...
Jabber:  ste...@re...
WWW:     http://www.reucon.com

Steuernummern 215/5140/1791 USt-IdNr. DE220701760

Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Julien T. <om...@gm...>]

Stefan Reuter wrote:
> I hope this makes things a bit clearer.

Thanks Stefan for the clear explanation. I did not realized it was as 
simple as setting a boolean in the mapping strategy object. Actually 
I've never really played with these strategy classes.

Thanks for the heads up!
Julien

Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls

[Johan S. <js...@sv...>]

Hi again, hi Stefan, !

Sorry for the delay but I've been strangled by deadlines and boring stuff.

Now posting my code, so our guru Stefan? might give some input on the problems I'm having.

The Answer-class below.service will get called for 7-8 simultanious calls. So 7-8 threads working at the same time if I've understood this.
In my code, each thread is doing AgiChannel-work, AsteriskManagerInterface-work, mysql-db-work, and playing/receiving dmtf-tones, and logging to one common logfile using log4J. All threads use the same logfile simultaneously too.

Recap of my problem. When checking my outputlog (not included), I see, sometimes one thread have log entries of playing dtmf-tones, and starting the getDTMF-function. Then other threads are executing in between, maybe one more call is answered or hungup.
Then the end of getDTMF-logentry is present (and no tones could be found on the channel).
It works perfectly when only one or two threads are working.

I suspected my cpu was overloaded with DTMF-detection, so I invested in the VPMADTO32 echo-cancelling/hw-dtmf-supported dsp-card for my analog 8ch TDM800P. But still the same problem. So dtmf-detection was not the issue I think. :(

I'm posting extracts of my Agi-code in java, and hopefully it will show fatal errors stopping performance to you, dear gurus. I apologize for stripping things, so it will not compile, but I must remove company-private stuff before posting in public.

Sincerely,
Johan
>From Sweden :)

My code
============


=========================
ANSWER CLASS
=========================

package com.test.johan.callManager;

import javax.jms.*;
import java.io.FileInputStream;
import java.io.IOException;
import java.sql.SQLException;
import java.text.DateFormat;
import java.util.Date;
import java.util.InvalidPropertiesFormatException;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

import javax.jms.JMSException;

import org.apache.log4j.Logger;
import org.asteriskjava.fastagi.AgiChannel;
import org.asteriskjava.fastagi.AgiException;
import org.asteriskjava.fastagi.AgiRequest;
import org.asteriskjava.fastagi.BaseAgiScript;
import org.asteriskjava.live.AsteriskChannel;
import org.asteriskjava.live.ChannelState;
import org.asteriskjava.live.DefaultAsteriskServer;
import org.asteriskjava.live.HangupCause;
import org.asteriskjava.live.ManagerCommunicationException;
import org.asteriskjava.live.NoSuchChannelException;

public class Answer extends BaseAgiScript {

  private static enum AnswerState {ALARM_INIT, ALARM_MSG_REQUEST, ALARM_GET_DTMFTONES, ALARM_CHECK_IF_LASTMESSAGE, ALARM_VERIFYCHECKSUM_AND_RESPOND, ALARM_HANDLE_MESSAGETYPE, ALARM_RETRY_BROKEN_MESSAGE, ALARM_PAUS_BEFORE_NEXTMESSAGE, ALARM_ENDOFREQUESTS,  ALARM_NODTMF_ERROR};


  private DefaultAsteriskServer asteriskServer;

  private DbApi dbApi;
  private static String propertyFile;
  private static Properties prop = new Properties();
  private static final Logger logger = Logger.getLogger(Answer.class);
  private static String answerInQueName;
  private static String answerOutQueName;
  private static String jmsUrl;
  private static String dbUrl;
  private static String dbUser;
  private static String dbPasswd;
  private static String asteriskUrl;
  private static String asteriskUser;
  private static String asteriskPasswd;
  private static int timeOutInSeconds;
  private static int PAUS_BETWEEN_DTMF_MESS = 500;

  public Answer() throws InvalidPropertiesFormatException, IOException {
    propertyFile = System.getProperty("com.johan.test.callManager.configFile");
    if (propertyFile == null) {
      propertyFile = "callManager.xml";
    }
    FileInputStream fis;
    fis = new FileInputStream(propertyFile);
    prop.loadFromXML(fis);
    //Read in all the properties from the file:
    answerInQueName = prop.getProperty("Answer.AnswerInQueName");
    answerOutQueName = prop.getProperty("Answer.AnswerOutQueName");
    jmsUrl = prop.getProperty("Answer.jmsUrl");
    dbUrl = prop.getProperty("General.dbUrl");
    dbUser = prop.getProperty("General.dbUser");
    dbPasswd = prop.getProperty("General.dbPasswd");
    asteriskUrl = prop.getProperty("General.asteriskUrl");
    asteriskUser = prop.getProperty("General.asteriskUser");
    asteriskPasswd = prop.getProperty("General.asteriskPasswd");
    timeOutInSeconds = Integer.parseInt(prop.getProperty("Answer.timeOutInSeconds"));
    numberOfRequestPerProtocol = Integer.parseInt(prop.getProperty("Answer.numberOfRequestPerProtocol"));

    dbApi = new DbApi(dbUrl, dbUser, dbPasswd);
    asteriskServer = new DefaultAsteriskServer(asteriskUrl, asteriskUser, asteriskPasswd);
    fis.close();
  }
  public void service(AgiRequest request, AgiChannel agiChannel){

    // Thread.currentThread().setPriority(1);

    JmsHandler jmsHandler = null;
    AsteriskChannel incomingChannel = null;
    Date timeOfAlarmCall = new Date();

    GenericMessage incomingMessage = null;
    int lineNumber = -1;
    boolean keepRequesting = true;
    String dtmfIncomingMessage="";
    AnswerState state = AnswerState.ALARM_INIT;
    AlarmSwitchMessageType messageType = null;

    // ########################################################################
        // MAIN WHILE
        while (keepRequesting && channelIsUp)
        {
          switch(state)
          {
              case ALARM_INIT:
              {
                logger.debug("ALARM_INIT");
                try
                {
                  logger.debug("Creating JmsHandler");
                  jmsHandler = new JmsHandler(jmsUrl, answerInQueName, answerOutQueName);
                  agiChannel.answer();
                  String incomingChannelName = agiChannel.getName();
                  timeOfAlarmCall = new Date();
                  logger.debug("Answered incoming call on channel: " + incomingChannelName + " on date: " + DateFormat.getDateInstance().format(timeOfAlarmCall));
                  // Wait a little, so channel is setup, before we send DTMF-tones.
                  Thread.sleep(500);

                  //We use incomingChannel instead of the incoming agiChannel because it supports redirect and playDtmf.
                  //In theory this could have been done by passing AgiActions to the channel.
                  incomingChannel = CallManager.getChannelFromAsteriskServer(incomingChannelName, asteriskServer);
                  state = AnswerState.ALARM_MSG_REQUEST;

                } catch (SQLException se)
                {
                  state = AnswerState.ALARM_ENDOFREQUESTS; // eller init-error...
                } catch(TestException se)
                {
                  state =AnswerState.ALARM_ENDOFREQUESTS; // eller init-error...
                } catch (InterruptedException ie)
                {
                  state =AnswerState.ALARM_ENDOFREQUESTS; // eller init-error...
                }catch(ManagerCommunicationException mce)
                {
                  state =AnswerState.ALARM_ENDOFREQUESTS; // eller init-error...
                }
                catch(AgiException ae)
                {
                  state = AnswerState.ALARM_ENDOFREQUESTS; // eller init-error...
                }
                catch( JMSException je)
                {
                  state =AnswerState.ALARM_ENDOFREQUESTS; // eller init-error...
                }
            }break;

            case ALARM_MSG_REQUEST:
            {
              logger.debug("ALARM_MSG_REQUEST");
              // send dtmf-tones and wait for the response dtmf-tones
              try
              {
                // Thread.currentThread().setPriority(10);
                playDtmf(TestProtocol.MESSAGE_REQUEST, incomingChannel);
                state = AnswerState.ALARM_GET_DTMFTONES;
              } catch (ManagerCommunicationException mce) {

                logger.fatal("ERROR "+mce.getMessage());
                state = AnswerState.ALARM_ENDOFREQUESTS;
              } catch(TestException se)
              {
                logger.fatal("ERROR "+se.getMessage());
                state = AnswerState.ALARM_ENDOFREQUESTS;
              }

            }
            break;

            case ALARM_GET_DTMFTONES:
            {
              logger.debug("ALARM_GET_DTMFTONES");
              dtmfIncomingMessage = "";
              try
              {
                dtmfIncomingMessage = CallManager.getDtmfMessage(agiChannel);
                logger.debug("DTMF->PC: " + dtmfIncomingMessage);

                if(dtmfIncomingMessage.length()>0)
                  state = AnswerState.ALARM_ENDOFREQUESTS;
                else
                {  // Nothing received
                  statusMessageRetries++;
                  logger.fatal("ERROR No DTMF-received! ");
                  state = AnswerState.ALARM_ENDOFREQUESTS;
                }
              } catch(AgiException ae)
              {
                logger.fatal("ERROR "+ae.getMessage());
                state = AnswerState.ALARM_ENDOFREQUESTS;
              } catch(TestException se)
              {
                logger.fatal("ERROR "+se.getMessage());
                state = AnswerState.ALARM_ENDOFREQUESTS;
              }
            }
            break;

            case ALARM_ENDOFREQUESTS:
            {
              // Thread.currentThread().setPriority(1);

              logger.debug("ALARM_ENDOFREQUESTS");

              agiChannel = null;
              incomingChannel = null;
              // Close jms, release resources. This is not happening if we get an agiexception due to code right now
              jmsHandler.cleanUpConnection();

            }break;

          }  // end switch(state)

        }  // end while
  }


}



=========================
=========================
=========================
CLASS CALLED CALLMANAGER
=========================
=========================
=========================


package com.johan.test.callManager;

import org.apache.log4j.Logger;
import org.asteriskjava.fastagi.AgiChannel;
import org.asteriskjava.fastagi.AgiException;
import org.asteriskjava.live.AsteriskChannel;
import org.asteriskjava.live.AsteriskServer;
import org.asteriskjava.live.HangupCause;
import org.asteriskjava.live.ManagerCommunicationException;
import org.asteriskjava.live.NoSuchChannelException;

public class CallManager {
  private static final Logger logger = Logger.getLogger(CallManager.class);
  private static final int requestTimeOutInSeconds = 5;
  private static final int timeOutForEachDigitInMilliSeconds = 500;
  private static final int timeBetweenTonesInMs = 140;
  private static final int lengthOfDTMFTone = 140;
  private static enum dtmfState {DTMF_INIT, DTMF_WAIT_CHAR, DTMF_STORE_CHAR, DTMF_FINISHED };

  public CallManager () {

  }

  protected static AsteriskChannel getChannelFromAsteriskServer (String channelName, AsteriskServer asteriskServer) throws ManagerCommunicationException {
    AsteriskChannel channel = null;
    for (AsteriskChannel asteriskChannel : asteriskServer.getChannels()) {
      // Checks if the name matches the incoming string:
      if ((asteriskChannel.getName().compareTo(channelName) == 0)) {
          channel = asteriskChannel;
      }
    }
    return channel;
  }

  protected static void playDtmf(String dtmfString, AsteriskChannel channel) throws ManagerCommunicationException,
  IllegalArgumentException, TestException {

    String sentString[] = new String[50];

    logger.debug("PC->DTMF:" + dtmfString);

    int length = dtmfString.length();
    for (int j = 0; j < length; j++)
    {
      try
      {
        channel.playDtmf(dtmfString.substring(j, j + 1));
        sentString[timerIndex] = dtmfString.substring(j, j + 1);
      }
      catch (NoSuchChannelException e)
      {
        // throw new TestException("Channel disconnected while playing Dtmf!");
      }
      try
      {
        Thread.sleep(timeBetweenTonesInMs);
      } catch (InterruptedException e)
      {
        // logger.fatal(e.getMessage());
      }
    }

  }

  protected static String getDtmfMessage(AgiChannel agiChannel) throws AgiException, TestException {
    // logger.debug("Starting getDtmfMessage()");
    long timerValue[] = new long[50];
    long timerValueBefore[] = new long[50];
    char receivedValue[] = new char[50];
    int timerIndex = 0;

    int cyclesBeforeTimeout = (1000 * requestTimeOutInSeconds) / timeOutForEachDigitInMilliSeconds;
    int i = 0;

    String message = "";
    char[] charBuf = new char[150];
    int charPos=0;
    char newChar = 0;
    long startTime = 0;
    boolean running = true;
    dtmfState state = dtmfState.DTMF_INIT;
    int waitDelay = 1000;

    while(running)
    {
      switch(state)
      {
        case DTMF_INIT:
        {
          startTime = System.currentTimeMillis();
          waitDelay = 10000;
          state = dtmfState.DTMF_WAIT_CHAR;
          logger.debug("Get-DTMF start");
        }
        break;

        case DTMF_WAIT_CHAR:
        {
          if(charPos>64)
            state = dtmfState.DTMF_FINISHED;
          try{
            if (agiChannel.getChannelStatus() == AlarmSwitch.AGI_CHANNEL_STATE_UP) {
              newChar = agiChannel.waitForDigit(waitDelay);
              if(newChar>0)
                state =  dtmfState.DTMF_STORE_CHAR;
              else
              {
                // logger.debug("No more DTMF-tones, timeout!");
                state = dtmfState.DTMF_FINISHED;
              }
            }
            else {
              throw new TestException("Channel disconnected while waiting for digits!");
            }
          } catch(AgiException ae)
          {
            // logger.fatal("ERROR " + ae.getMessage());
            state = dtmfState.DTMF_FINISHED;
          }
        }
        break;

        case DTMF_STORE_CHAR:
        {
          message = message + String.valueOf(newChar);
          //charBuf[charPos]  = newChar;
          //charPos++;
          waitDelay = 400;
          state = dtmfState.DTMF_WAIT_CHAR;
        }break;

        case DTMF_FINISHED:
        {
          logger.debug("Get-DTMF done");
          running = false;
        }break;
      }
    }

    // message = new String(charBuf);
    return message;
  }

}




-----Ursprungligt meddelande-----
Från: ast...@li... [mailto:ast...@li...] För Stefan Reuter
Skickat: den 14 februari 2008 16:32
Till: ast...@li...
Ämne: [SPAM] - Re: [Asterisk-java-users] Problems with concurrent java-threads answering incoming calls - Email found in subject

Johan Sandgren wrote:
> What about my problem, any clever ideas?

would you mind to post your code?

=Stefan

--
reuter network consulting
Neusser Str. 110
50760 Koeln
Germany
Telefon: +49 221 1305699-0
Telefax: +49 221 1305699-90
E-Mail:  ste...@re...
Jabber:  ste...@re...
WWW:     http://www.reucon.com

Steuernummern 215/5140/1791 USt-IdNr. DE220701760