self is perfectly fine
self
self to create a child and forward the message
var self = Self;
_watcher.Created += (sender, e) =>
{
var message = new MyMessage(filePath);
self.Tell(message, ActorRefs.NoSender);
};Self is just Context.Self
var self = Self Self is showing up as a NotSupportedException with this:System.NotSupportedException: There is no active ActorContext, this is most likely due to use of async operations from within this actor.
at Akka.Actor.ActorBase.get_Context()
at Akka.Actor.ActorBase.get_Self()self shows up as the right actor. But when it never receives the message it tells itself.
Nope, this is basically what is running, minus some logging:
private void Active()
{
Receive<MyMessage>(message =>
{
TellNewChild(message);
});
_watcher = new FileSystemWatcher("directory", "*.csv")
{
NotifyFilter = NotifyFilters.LastAccess
| NotifyFilters.LastWrite
| NotifyFilters.FileName
| NotifyFilters.DirectoryName,
EnableRaisingEvents = true
};
var self = Self;
_watcher.Created += (sender, e) =>
{
var message = new MyMessage(e.FullPath;
self.Tell(message, ActorRefs.NoSender);
};
}But it never hits the MyMessage receive handler. It is in this Active behavior.
Have a question about killing actors. Is there a way to kill an actor, including the current message its processing? From looking at the docs, it appears that both the shutdown and kill functions send a message to the actor that would get processed after the current message.
we have a situation where sometimes some code within the actor can hang and block, so would like to be able to restart the actor from above. Or do we need to add timeouts etc to the actors message processing code?
@Greatsamps as you've probably noticed, an actor's Receive method is just a block of synchronous code. Once it started executing, it will run until completion. There's not much we can do here, as only operating system can interrupt code running on a given thread, and OS itself is not aware of existence of actors.
Usually the reason behind having long running blocking code is either trying to do synchronous I/O (which you should never do, pretty much every .NET API now exposes Async API for that) or long running loop. For the latter case one way is to split long running process into sequence of steps or batches and send them as operations to Self, eg. instead of having:
Receive<Msg>(msg => {
var sum = 0;
for (var i = 0; i < msg.Count; i++) {
sum += DoSomething(i);
}
Sender.Tell(sum);
});You can model it as:
Receive<Msg>(msg => {
var sum = msg.State;
// process message in batches of 1000 elements
var until = Math.Min(msg.From + 1000, msg.Count);
var i = msg.From;
for (; i < until; i++) {
sum += DoSomething(i);
}
if (i < msg.Count) {
Self.Forward(new Msg(state: sum, from: i, count: msg.Count));
} else {
Sender.Tell(sum);
}
});
I've encountered this error whilst using Akka.Persistence.MongoDb and Akka.Cluster.Sharding
Exception in ReceiveRecover when replaying event type ["Akka.Cluster.Sharding.PersistentShardCoordinator+ShardRegionRegistered"] with sequence number [1] for persistenceId ["/system/sharding/TestCoordinator/singleton/coordinator"]
System.ArgumentNullException: Value cannot be null.
Parameter name: key
at System.Collections.Immutable.Requires.FailArgumentNullException(String parameterName)
at System.Collections.Immutable.ImmutableDictionary`2.ContainsKey(TKey key)
at Akka.Cluster.Sharding.PersistentShardCoordinator.State.Updated(IDomainEvent e)
at Akka.Cluster.Sharding.PersistentShardCoordinator.ReceiveRecover(Object message)
at Akka.Actor.ActorBase.AroundReceive(Receive receive, Object message)
at Akka.Persistence.Eventsourced.<>cDisplayClass92_0.<Recovering>b1(Receive receive, Object message)
This will only occur when state-store-mode is set to persistence and after the cluster has been created once. It seems that the ShardRegionRegistered object is unable to be serialized correctly (See below), presumably due to the fact that it contains an IActorRef.
This does not happen when state-store-mode is set to ddata, however I would ideally prefer to use remember-entities which I cannot get to work with ddata.
{
"_id" : "/system/sharding/TestCoordinator/singleton/coordinator_4",
"PersistenceId" : "/system/sharding/TestCoordinator/singleton/coordinator",
"SequenceNr" : NumberLong(4),
"IsDeleted" : false,
"Payload" : {
"_t" : "Akka.Cluster.Sharding.PersistentShardCoordinator+ShardRegionRegistered, Akka.Cluster.Sharding"
},
"Manifest" : "Akka.Persistence.Persistent, Akka.Persistence",
"Ordering" : Timestamp(1568893010, 1),
"Tags" : null,
"SerializerId" : null
}
@Greatsamps dotnet has a feature Thread.Abort(), but its a rabbit hole and i dont know if its still supported in dotnet core. If there is really a situation where it a Thread.Abort seams required it is near always to restart the whole process else you can get so many bad side effects. If you realy need to abort some process and dont have any other option like CancellationToken or a timeout paramter, then u can just create a deticated Thread/Task inside the actor and Abort it from the actor thread there.
Thread taskThread = null;
var task = Task.Run(() => {
taskThread = Thread.CurrentThread;
Thread.Sleep(Timeout.Infinite);
});
Thread.Sleep(2000); //wait to start the new thread
taskThread.Abort();But it should realy only be used for a 3th libery API where there is no other option to cancel the process. to make a GC.Collect() after the abort is a good idea too.
IAsyncLifetime fixtures to keep Docker containers alive while the entire test suite runs
