You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nlpcraft.apache.org by se...@apache.org on 2020/09/11 10:14:16 UTC
[incubator-nlpcraft] branch NLPCRAFT-41-config created (now 6ef5588)
This is an automated email from the ASF dual-hosted git repository.
sergeykamov pushed a change to branch NLPCRAFT-41-config
in repository https://gitbox.apache.org/repos/asf/incubator-nlpcraft.git.
at 6ef5588 WIP.
This branch includes the following new commits:
new 6ef5588 WIP.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-nlpcraft] 01/01: WIP.
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sergeykamov pushed a commit to branch NLPCRAFT-41-config
in repository https://gitbox.apache.org/repos/asf/incubator-nlpcraft.git
commit 6ef5588335d1bbc4222d6e77f53bef427ba2d651
Author: Sergey Kamov <se...@apache.org>
AuthorDate: Fri Sep 11 13:14:07 2020 +0300
WIP.
---
nlpcraft/src/main/resources/nlpcraft.conf | 5 +-
.../org/apache/nlpcraft/model/NCModelView.java | 31 ++++++
.../nlpcraft/probe/mgrs/cmd/NCCommandManager.scala | 2 +-
.../conversation/NCConversationDescriptor.scala | 26 +++--
.../mgrs/conversation/NCConversationManager.scala | 54 +++++++---
.../probe/mgrs/deploy/NCDeployManager.scala | 110 +++++++++++----------
.../mgrs/dialogflow/NCDialogFlowManager.scala | 76 ++++++++------
.../nlpcraft/probe/mgrs/model/NCModelManager.scala | 5 +-
.../probe/mgrs/nlp/NCProbeEnrichmentManager.scala | 2 +-
9 files changed, 194 insertions(+), 117 deletions(-)
diff --git a/nlpcraft/src/main/resources/nlpcraft.conf b/nlpcraft/src/main/resources/nlpcraft.conf
index 7ef1e8f..bd2225d 100644
--- a/nlpcraft/src/main/resources/nlpcraft.conf
+++ b/nlpcraft/src/main/resources/nlpcraft.conf
@@ -310,8 +310,9 @@ nlpcraft {
# When exceeded the request will be automatically rejected.
resultMaxSizeBytes = 1048576
- // TODO:
- synonymsWarnValue = 10000
+ // TODO: these values shouldn't be too big, otherwise model dependt timeouts have to sense.
+ conversation.check.period.secs = 60
+ dialog.check.period.secs = 60
}
# Basic NLP toolkit to use on both server and probes. Possible values:
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/model/NCModelView.java b/nlpcraft/src/main/scala/org/apache/nlpcraft/model/NCModelView.java
index a0dcf3c..8fdee78 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/model/NCModelView.java
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/model/NCModelView.java
@@ -17,6 +17,7 @@
package org.apache.nlpcraft.model;
+import java.time.Duration;
import java.util.*;
/**
@@ -37,6 +38,36 @@ public interface NCModelView extends NCMetadata {
*/
int DFLT_JIGGLE_FACTOR = 2;
+ // TODO: comments
+ int DFLT_SUSP_MANY_SYNONYMS = 1000;
+ boolean DFLT_SUSP_MANY_SYNONYMS_ERROR = false;
+
+ Duration DFLT_CONV_USAGE_TIMEOUT = Duration.ofMinutes(60);
+ Duration DFLT_CONV_UPDATE_TIMEOUT = Duration.ofMinutes(5);
+ int DFLT_CONV_MAX_DEPTH = 3;
+
+ Duration DFLT_DIALOG_TIMEOUT = Duration.ofMinutes(60);
+
+ // TODO: comments, review and move to the end of file.
+ default int getSuspManySynonyms() {
+ return DFLT_SUSP_MANY_SYNONYMS;
+ }
+ default boolean isSuspManySynonymsError() {
+ return DFLT_SUSP_MANY_SYNONYMS_ERROR;
+ }
+ default Duration getConvUsageTimeout() {
+ return DFLT_CONV_USAGE_TIMEOUT;
+ }
+ default Duration getConvUpdateTimeout() {
+ return DFLT_CONV_UPDATE_TIMEOUT;
+ }
+ default int getConvMaxDepth() {
+ return DFLT_CONV_MAX_DEPTH;
+ }
+ default Duration getDialogTimeout() {
+ return DFLT_DIALOG_TIMEOUT;
+ }
+
/**
* Default value returned from {@link #getJiggleFactor()} method.
*/
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/cmd/NCCommandManager.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/cmd/NCCommandManager.scala
index f53aa6e..bd9d44a 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/cmd/NCCommandManager.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/cmd/NCCommandManager.scala
@@ -108,7 +108,7 @@ object NCCommandManager extends NCService {
case "S2P_MODEL_INFO" ⇒
val mdlId = msg.data[String]("mdlId")
- val mdlData = NCModelManager.getModelData(mdlId)
+ val mdlData = NCModelManager.getModelData(mdlId).getOrElse(throw new NCE(s"Model not found: $mdlId"))
val macros = mdlData.model.getMacros.asInstanceOf[Serializable]
val syns = mdlData.model.getElements.asScala.
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationDescriptor.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationDescriptor.scala
index 7fad7f7..dbee138 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationDescriptor.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationDescriptor.scala
@@ -29,19 +29,23 @@ import org.apache.nlpcraft.model._
import scala.collection.JavaConverters._
import scala.collection.mutable
-import scala.concurrent.duration._
/**
* Conversation as an ordered set of utterances.
*/
-case class NCConversationDescriptor(usrId: Long, mdlId: String) extends LazyLogging with NCOpenCensusTrace {
+case class NCConversationDescriptor(
+ usrId: Long,
+ mdlId: String,
+ updateTimeoutMs: Long,
+ maxDepth: Int
+) extends LazyLogging with NCOpenCensusTrace {
/**
*
* @param token
* @param tokenTypeUsageTime
*/
case class TokenHolder(token: NCToken, var tokenTypeUsageTime: Long = 0)
-
+
/**
*
* @param holders Tokens holders.
@@ -49,12 +53,6 @@ case class NCConversationDescriptor(usrId: Long, mdlId: String) extends LazyLogg
* @param tstamp Request timestamp. Used just for logging.
*/
case class ConversationItem(holders: mutable.ArrayBuffer[TokenHolder], srvReqId: String, tstamp: Long)
-
- // After 5 mins pause between questions we clear the STM.
- private final val CONV_CLEAR_DELAY = 5.minutes.toMillis
-
- // If token is not used in last 3 requests, it is removed from the conversation.
- private final val MAX_DEPTH = 3
// Short-Term-Memory.
private val stm = mutable.ArrayBuffer.empty[ConversationItem]
@@ -83,7 +81,7 @@ case class NCConversationDescriptor(usrId: Long, mdlId: String) extends LazyLogg
attempt += 1
// Conversation cleared by timeout or when there are too much unsuccessful requests.
- if (now - lastUpdateTstamp > CONV_CLEAR_DELAY) {
+ if (now - lastUpdateTstamp > updateTimeoutMs) {
stm.clear()
logger.info(s"Conversation reset by timeout [" +
@@ -91,16 +89,16 @@ case class NCConversationDescriptor(usrId: Long, mdlId: String) extends LazyLogg
s"mdlId=$mdlId" +
s"]")
}
- else if (attempt > MAX_DEPTH) {
+ else if (attempt > maxDepth) {
stm.clear()
-
+
logger.info(s"Conversation reset after too many unsuccessful requests [" +
s"usrId=$usrId, " +
s"mdlId=$mdlId" +
s"]")
}
else {
- val minUsageTime = now - CONV_CLEAR_DELAY
+ val minUsageTime = now - updateTimeoutMs
val toks = lastToks.flatten
for (item ← stm) {
@@ -177,7 +175,7 @@ case class NCConversationDescriptor(usrId: Long, mdlId: String) extends LazyLogg
// Last used tokens processing.
lastToks += toks
- val delCnt = lastToks.length - MAX_DEPTH
+ val delCnt = lastToks.length - maxDepth
if (delCnt > 0)
lastToks.remove(0, delCnt)
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationManager.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationManager.scala
index 52c1fb8..aea99a4 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationManager.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/conversation/NCConversationManager.scala
@@ -21,9 +21,11 @@ import java.util.concurrent.{Executors, ScheduledExecutorService, TimeUnit}
import io.opencensus.trace.Span
import org.apache.nlpcraft.common._
+import org.apache.nlpcraft.common.config.NCConfigurable
+import org.apache.nlpcraft.probe.mgrs.model.NCModelManager
import scala.collection._
-import scala.concurrent.duration._
+import scala.collection.mutable.ArrayBuffer
/**
* Conversation manager.
@@ -32,21 +34,26 @@ object NCConversationManager extends NCService {
case class Key(userId: Long, mdlId: String)
case class Value(conv: NCConversationDescriptor, var tstamp: Long = 0)
- // Check frequency and timeout.
- private final val CHECK_PERIOD = 5.minutes.toMillis
- private final val TIMEOUT = 1.hour.toMillis
+ private object Config extends NCConfigurable {
+ def periodMs: Long = getInt(s"nlpcraft.probe.conversation.check.period.secs") * 1000
- @volatile private var convs: mutable.Map[Key, Value] = _
+ def check(): Unit =
+ if (periodMs <= 0)
+ abortWith(s"Value of 'nlpcraft.probe.conversation.check.period.secs' must be positive")
+ }
+
+ Config.check()
+ @volatile private var convs: mutable.Map[Key, Value] = _
@volatile private var gc: ScheduledExecutorService = _
override def start(parent: Span = null): NCService = startScopedSpan("start", parent) { _ ⇒
gc = Executors.newSingleThreadScheduledExecutor
convs = mutable.HashMap.empty[Key, Value]
- gc.scheduleWithFixedDelay(() ⇒ clearForTimeout(), CHECK_PERIOD, CHECK_PERIOD, TimeUnit.MILLISECONDS)
+ gc.scheduleWithFixedDelay(() ⇒ clearForTimeout(), Config.periodMs, Config.periodMs, TimeUnit.MILLISECONDS)
- logger.info(s"Conversation manager GC started [checkPeriodMs=$CHECK_PERIOD, timeoutMs=$TIMEOUT]")
+ logger.info(s"Conversation manager GC started [checkPeriodMs=${Config.periodMs}]")
super.start()
}
@@ -62,15 +69,27 @@ object NCConversationManager extends NCService {
/**
*
*/
- private def clearForTimeout(): Unit = {
- val ms = U.nowUtcMs() - TIMEOUT
-
- startScopedSpan("clear", "checkPeriodMs" → CHECK_PERIOD, "timeoutMs" → TIMEOUT) { _ ⇒
- convs.synchronized {
- convs --= convs.filter(_._2.tstamp < ms).keySet
+ private def clearForTimeout(): Unit =
+ startScopedSpan("clearForTimeout", "checkPeriodMs" → Config.periodMs) { _ ⇒
+ try
+ convs.synchronized {
+ val delKeys = ArrayBuffer.empty[Key]
+
+ for ((key, value) ← convs)
+ NCModelManager.getModelData(key.mdlId) match {
+ case Some(data) ⇒
+ if (value.tstamp < System.currentTimeMillis() -data.model.getConvUsageTimeout.toMillis)
+ delKeys += key
+ case None ⇒ delKeys += key
+ }
+
+ convs --= delKeys
+ }
+ catch {
+ case e: Throwable ⇒ logger.error("Clean method unexpected error", e)
}
+
}
- }
/**
* Gets conversation for given key.
@@ -81,8 +100,13 @@ object NCConversationManager extends NCService {
*/
def getConversation(usrId: Long, mdlId: String, parent: Span = null): NCConversationDescriptor =
startScopedSpan("getConversation", parent, "usrId" → usrId, "modelId" → mdlId) { _ ⇒
+ val mdl = NCModelManager.getModelData(mdlId).getOrElse(throw new NCE(s"Model not found: $mdlId")).model
+
convs.synchronized {
- val v = convs.getOrElseUpdate(Key(usrId, mdlId), Value(NCConversationDescriptor(usrId, mdlId)))
+ val v = convs.getOrElseUpdate(
+ Key(usrId, mdlId),
+ Value(NCConversationDescriptor(usrId, mdlId, mdl.getConvUpdateTimeout.toMillis, mdl.getConvMaxDepth))
+ )
v.tstamp = U.nowUtcMs()
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/deploy/NCDeployManager.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/deploy/NCDeployManager.scala
index 20837a0..071e308 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/deploy/NCDeployManager.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/deploy/NCDeployManager.scala
@@ -93,9 +93,6 @@ object NCDeployManager extends NCService with DecorateAsScala {
def model: Option[String] = getStringOpt(s"$pre.model")
def models: Seq[String] = getStringList(s"$pre.models")
def jarsFolder: Option[String] = getStringOpt(s"$pre.jarsFolder")
-
- // TODO: property name.
- def synonymsWarnValue: Int = getInt(s"$pre.synonymsWarnValue")
}
/**
@@ -132,51 +129,14 @@ object NCDeployManager extends NCService with DecorateAsScala {
private def wrap(mdl: NCModel): NCModelData = {
require(mdl != null)
+ checkModelConfig(mdl)
+
val mdlId = mdl.getId
- val mdlName = mdl.getName
- val mdlVer = mdl.getVersion
-
- // Verify models' identities.
-
- if (mdlId == null)
- throw new NCE(s"Model ID is not provided: $mdlName")
- if (mdlName == null)
- throw new NCE(s"Model name is not provided: $mdlId")
- if (mdlVer == null)
- throw new NCE(s"Model version is not provided: $mdlId")
- if (mdlName != null && mdlName.isEmpty)
- throw new NCE(s"Model name cannot be empty string: $mdlId")
- if (mdlId != null && mdlId.isEmpty)
- throw new NCE(s"Model ID cannot be empty string: $mdlId")
- if (mdlVer != null && mdlVer.length > 16)
- throw new NCE(s"Model version cannot be empty string: $mdlId")
- if (mdlName != null && mdlName.length > 64)
- throw new NCE(s"Model name is too long (64 max): $mdlId")
- if (mdlId != null && mdlId.length > 32)
- throw new NCE(s"Model ID is too long (32 max): $mdlId")
- if (mdlVer != null && mdlVer.length > 16)
- throw new NCE(s"Model version is too long (16 max): $mdlId")
for (elm ← mdl.getElements.asScala)
if (!elm.getId.matches(ID_REGEX))
throw new NCE(s"Model element ID '${elm.getId}' does not match '$ID_REGEX' regex in: $mdlId")
- @throws[NCE]
- def checkCollection(name: String, col: Any): Unit =
- if (col == null)
- throw new NCE(s"Collection can be empty but cannot be null [modelId=$mdlId, name=$name]")
-
- checkCollection("additionalStopWords", mdl.getAdditionalStopWords)
- checkCollection("elements", mdl.getElements)
- checkCollection("enabledBuiltInTokens", mdl.getEnabledBuiltInTokens)
- checkCollection("excludedStopWords", mdl.getExcludedStopWords)
- checkCollection("parsers", mdl.getParsers)
- checkCollection("suspiciousWords", mdl.getSuspiciousWords)
- checkCollection("macros", mdl.getMacros)
- checkCollection("metadata", mdl.getMetadata)
-
- checkModelConfig(mdl)
-
val allSyns = mdl.getElements.asScala.flatMap(_.getSynonyms.asScala)
mdl.getMacros.asScala.keys.foreach(makro ⇒
@@ -816,11 +776,36 @@ object NCDeployManager extends NCService with DecorateAsScala {
* @param mdl Model.
*/
private def checkModelConfig(mdl: NCModel): Unit = {
+ val mdlId = mdl.getId
+
+ @throws[NCE]
+ def checkMandatoryString(value: String, name: String, maxLen: Int): Unit =
+ if (value == null)
+ throw new NCE(s"$name is not provided [modeId=$mdlId]")
+ else if (value.isEmpty)
+ throw new NCE(s"$name cannot be empty string [modeId=$mdlId]")
+ else if (value.length > maxLen)
+ throw new NCE(s"$name is too long ($maxLen max): $value [modeId=$mdlId]")
+
+ checkMandatoryString(mdl.getId, "Model ID", 32)
+ checkMandatoryString(mdl.getName, "Model name", 64)
+ checkMandatoryString(mdl.getVersion, "Model version", 16)
+
+ @throws[NCE]
+ def checkNotNull(value: AnyRef, name: String): Unit =
+ if (value == null)
+ throw new NCE(s"$name is not provided [modeId=$mdlId]")
+
+ checkNotNull(mdl.getConvUsageTimeout, "Conversation usage timeout")
+ checkNotNull(mdl.getConvUpdateTimeout, "Conversation update timeout")
+ checkNotNull(mdl.getDialogTimeout, "Dialog timeout")
+
+ @throws[NCE]
def checkInt(v: Int, name: String, min: Int = 0, max: Int = Integer.MAX_VALUE): Unit =
if (v < min)
- throw new NCE(s"Invalid model configuration value '$name' [value=$v, min=$min], modelId: ${mdl.getId}.")
+ throw new NCE(s"Invalid model configuration value '$name' [value=$v, min=$min, modelId=$mdlId]")
else if (v > max)
- throw new NCE(s"Invalid model configuration value '$name' [value=$v, max=$min], modelId: ${mdl.getId}.")
+ throw new NCE(s"Invalid model configuration value '$name' [value=$v, max=$min, modelId=$mdlId]")
checkInt(mdl.getMaxUnknownWords, "maxUnknownWords")
checkInt(mdl.getMaxFreeWords, "maxFreeWords")
@@ -831,18 +816,34 @@ object NCDeployManager extends NCService with DecorateAsScala {
checkInt(mdl.getMaxTokens, "maxTokens", max = 100)
checkInt(mdl.getMaxWords, "maxWords", min = 1, max = 100)
checkInt(mdl.getJiggleFactor, "jiggleFactor", max = 4)
+ checkInt(mdl.getSuspManySynonyms, "suspManySynonyms", min = 1)
+ checkInt(mdl.getConvMaxDepth, "convMaxDepth", min = 1)
+
+ @throws[NCE]
+ def checkCollection(name: String, col: Any): Unit =
+ if (col == null)
+ throw new NCE(s"Collection can be empty but cannot be null [modelId=$mdlId, name=$name]")
+
+ checkCollection("additionalStopWords", mdl.getAdditionalStopWords)
+ checkCollection("elements", mdl.getElements)
+ checkCollection("enabledBuiltInTokens", mdl.getEnabledBuiltInTokens)
+ checkCollection("excludedStopWords", mdl.getExcludedStopWords)
+ checkCollection("parsers", mdl.getParsers)
+ checkCollection("suspiciousWords", mdl.getSuspiciousWords)
+ checkCollection("macros", mdl.getMacros)
+ checkCollection("metadata", mdl.getMetadata)
val unsToks =
mdl.getEnabledBuiltInTokens.asScala.filter(t ⇒
// 'stanford', 'google', 'opennlp', 'spacy' - any names, not validated.
t == null ||
- !TOKENS_PROVIDERS_PREFIXES.exists(typ ⇒ t.startsWith(typ)) ||
- // 'nlpcraft' names validated.
- (t.startsWith("nlpcraft:") && !NCModelView.DFLT_ENABLED_BUILTIN_TOKENS.contains(t))
+ !TOKENS_PROVIDERS_PREFIXES.exists(typ ⇒ t.startsWith(typ)) ||
+ // 'nlpcraft' names validated.
+ (t.startsWith("nlpcraft:") && !NCModelView.DFLT_ENABLED_BUILTIN_TOKENS.contains(t))
)
if (unsToks.nonEmpty)
- throw new NCE(s"Invalid model 'enabledBuiltInTokens' token IDs: ${unsToks.mkString(", ")}, modelId: ${mdl.getId}.")
+ throw new NCE(s"Invalid model 'enabledBuiltInTokens' token IDs: ${unsToks.mkString(", ")} [modelId=${mdl.getId}]")
}
/**
@@ -935,11 +936,16 @@ object NCDeployManager extends NCService with DecorateAsScala {
if (size == 0)
logger.warn(s"Element '$elemId' doesn't have synonyms [modelId=$mdlId]")
- else if (size > Config.synonymsWarnValue)
- logger.warn(
+ else if (size > mdl.getSuspManySynonyms) {
+ val msg =
s"Element '$elemId' has too many ($size) synonyms. " +
- s"Make sure this is truly necessary [modelId=$mdlId]"
- )
+ s"Make sure this is truly necessary [modelId=$mdlId]"
+
+ if (mdl.isSuspManySynonymsError)
+ throw new NCE(msg)
+ else
+ logger.warn(msg)
+ }
val others = mdlSyns.filter {
case (otherId, _) ⇒ otherId != elemId
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/dialogflow/NCDialogFlowManager.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/dialogflow/NCDialogFlowManager.scala
index f04cbbe..b18a57e 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/dialogflow/NCDialogFlowManager.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/dialogflow/NCDialogFlowManager.scala
@@ -20,12 +20,12 @@ package org.apache.nlpcraft.probe.mgrs.dialogflow
import java.util.concurrent.{Executors, ScheduledExecutorService, TimeUnit}
import io.opencensus.trace.Span
-import org.apache.nlpcraft.common.NCService
-import org.apache.nlpcraft.common._
+import org.apache.nlpcraft.common.config.NCConfigurable
+import org.apache.nlpcraft.common.{NCService, _}
+import org.apache.nlpcraft.probe.mgrs.model.NCModelManager
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.duration._
/**
* Dialog flow manager.
@@ -34,33 +34,38 @@ object NCDialogFlowManager extends NCService {
case class Key(usrId: Long, mdlId: String)
case class Value(intent: String, tstamp: Long)
+ private object Config extends NCConfigurable {
+ def periodMs: Long = getInt(s"nlpcraft.probe.dialog.check.period.secs") * 1000
+
+ def check(): Unit =
+ if (periodMs <= 0)
+ abortWith(s"Value of 'nlpcraft.probe.conversation.check.period.secs' must be positive")
+ }
+
+ Config.check()
+
@volatile private var flow: mutable.Map[Key, ArrayBuffer[Value]] = _
-
- // Check frequency and timeout.
- private final val CHECK_PERIOD = 5.minutes.toMillis
- private final val TIMEOUT = 1.hour.toMillis
-
@volatile private var gc: ScheduledExecutorService = _
-
+
override def start(parent: Span = null): NCService = startScopedSpan("start", parent) { _ ⇒
flow = mutable.HashMap.empty[Key, ArrayBuffer[Value]]
gc = Executors.newSingleThreadScheduledExecutor
-
- gc.scheduleWithFixedDelay(() ⇒ clearForTimeout(), CHECK_PERIOD, CHECK_PERIOD, TimeUnit.MILLISECONDS)
-
- logger.info(s"Dialog flow manager GC started [checkPeriodMs=$CHECK_PERIOD, timeoutMs=$TIMEOUT]")
-
+
+ gc.scheduleWithFixedDelay(() ⇒ clearForTimeout(), Config.periodMs, Config.periodMs, TimeUnit.MILLISECONDS)
+
+ logger.info(s"Dialog flow manager GC started [checkPeriodMs=${Config.periodMs}]")
+
super.start()
}
-
+
override def stop(parent: Span = null): Unit = startScopedSpan("stop", parent) { _ ⇒
U.shutdownPools(gc)
-
+
logger.info("Dialog flow manager GC stopped")
-
+
super.stop()
}
-
+
/**
* Adds matched (winning) intent to the dialog flow for the given user and model IDs.
*
@@ -75,11 +80,11 @@ object NCDialogFlowManager extends NCService {
Value(intent, System.currentTimeMillis())
)
}
-
+
logger.trace(s"Added to dialog flow [intent=$intent, userId=$usrId, modelId=$mdlId]")
}
}
-
+
/**
* Gets sequence of intent ID sorted from oldest to newest (i.e. dialog flow) for given user and model IDs.
*
@@ -94,19 +99,32 @@ object NCDialogFlowManager extends NCService {
}
}
}
-
+
/**
- *
+ *
*/
- private def clearForTimeout(): Unit = {
- val ms = U.nowUtcMs() - TIMEOUT
-
- startScopedSpan("clearForTimeout", "checkPeriodMs" → CHECK_PERIOD, "timeoutMs" → TIMEOUT) { _ ⇒
- flow.synchronized {
- flow.values.foreach(arr ⇒ arr --= arr.filter(_.tstamp < ms))
+ private def clearForTimeout(): Unit =
+ startScopedSpan("clearForTimeout", "checkPeriodMs" → Config.periodMs) { _ ⇒
+ try
+ flow.synchronized {
+ val delKeys = ArrayBuffer.empty[Key]
+
+ for ((key, values) ← flow)
+ NCModelManager.getModelData(key.mdlId) match {
+ case Some(data) ⇒
+ val ms = System.currentTimeMillis() -data.model.getDialogTimeout.toMillis
+
+ values --= values.filter(_.tstamp < ms)
+
+ case None ⇒ delKeys += key
+ }
+
+ flow --= delKeys
+ }
+ catch {
+ case e: Throwable ⇒ logger.error("Clean method unexpected error", e)
}
}
- }
/**
* Clears dialog for given user and model IDs.
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/model/NCModelManager.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/model/NCModelManager.scala
index 106e822..f239ac0 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/model/NCModelManager.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/model/NCModelManager.scala
@@ -112,10 +112,9 @@ object NCModelManager extends NCService with DecorateAsScala {
/**
*
* @param mdlId Model ID.
- * @return
*/
- def getModelData(mdlId: String, parent: Span = null): NCModelData =
+ def getModelData(mdlId: String, parent: Span = null): Option[NCModelData] =
startScopedSpan("getModel", parent, "modelId" → mdlId) { _ ⇒
- mux.synchronized { data.getOrElse(mdlId, throw new NCE(s"Model not found: $mdlId")) }
+ mux.synchronized { data.get(mdlId) }
}
}
\ No newline at end of file
diff --git a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/nlp/NCProbeEnrichmentManager.scala b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/nlp/NCProbeEnrichmentManager.scala
index f30e729..f1ff8c1 100644
--- a/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/nlp/NCProbeEnrichmentManager.scala
+++ b/nlpcraft/src/main/scala/org/apache/nlpcraft/probe/mgrs/nlp/NCProbeEnrichmentManager.scala
@@ -315,7 +315,7 @@ object NCProbeEnrichmentManager extends NCService with NCOpenCensusModelStats {
logger.info(s"REJECT response $msgName sent [srvReqId=$srvReqId, response=${errMsg.get}]")
}
- val mdlData = NCModelManager.getModelData(mdlId, span)
+ val mdlData = NCModelManager.getModelData(mdlId, span).getOrElse(throw new NCE(s"Model not found: $mdlId"))
var errData: Option[(String, Int)] = None