akkadotnet/akka.net

Port of Akka actors for .NET

Arjen Smits

@Danthar

maar vanwege last minute venue change is het niet door gegaan

Arjen Smits

@Danthar

Maargoed, uiteindelijk was het een stuk tammer geworden dan geplanned. Bij de omschrijving was gezet dat het voor mensen was die al bekend waren met de basics van Akka. Maar toen ik voor de groep stond had niemand ook maar de hello-world van akka gedaan :+

dus toen heb ik on-the-fly maar wat wijzigingen gemaakt ^^

AndreSteenbergen

@AndreSteenbergen

hummm (Engligh would be better for our gitter friends, sorry for starting in Dutch), Advanced Akka (java) is quite easy too find. More advanced stuff akka.net are more introductions and a reference to read up on more advanced stuff

basically you were forced by the audience to do the same :)

Sam13

@Sam13

Hello everybody. I have an actor which can be queried for files on his local filesystem. It will open the desired file and send it to the requesting actor.
Example:

private void OnReceive(FileTransferStart fileTransferStart)
 {
    var fileInfo = new FileInfo(Path.Combine(m_directory, fileTransferStart.FileName));
    FileIO.FromFile(fileInfo).Via(Flow.Create<ByteString>().Select(x => x.ToArray())).To(Sink.ActorRef<byte[]>(Sender, new FileTransferFinished())).Run(Context.Materializer()).ContinueWith(x =>
    {
        if (!x.Result.WasSuccessful)
        {
            Logger.ErrorException("Error during file transfer", x.Result.Error);
        }
    });
 }

Now I want to limit the number of active file transfers to some number, e.g. 25.
Can I use the Throttle functionality for that or do I have to implement something myself?
If throttle can be used, what are the parameters? In the source code I did not find any documentation which is understandable for me ;-).
Thanks in advance

Arjen Smits

@Danthar

your stream is now scoped to a single file

So using Throttle will only work on that file. The FileIO returns a bytestring per line i think.

If you want to throttle the amount of work currently being handled. you would have to handle that at a higher level then in your example

Arjen Smits

@Danthar

@Sam13 how you could do this with streams is by using a queue source, in which you enqueue your FileTransferStart requests, and connect your file processing flow to that Source.

Then put something in between to limit execution

as in, between the Source and your flow.

You might even do something really fancy and use a bidiflow, to build something custom

But a Throttle might be simple enough. There is plenty of docs available online about akka streams Throttle

matneyx

@matneyx

Apparently today is the day for file IO questions.

I have a List of filenames... I'd like to spin up a new child per filename, then ask the child how many lines the particular file has, and then I'd like to add the line-counts of all the filenames to another list.

Looking at the Child Per Entity Pattern (https://gigi.nullneuron.net/gigilabs/child-per-entity-pattern-in-akka-net/), I feel like I can foreach filename, generate my child, and that's where I'm kinda stuck... If I Ask instead of Tell (as he uses Tell in the example) I can't realistically bundle it into a task list, can I? A task list can't really yield results, as far as I'm aware...

I've also tried a ForEachAsync (https://stackoverflow.com/a/42570099/1672020) idea, but the Context doesn't work inside that Action.

I'm assuming there's a better solution for what I wanna do, but I don't have the slightest on what that is.

matneyx

@matneyx

TLDR; How can I: ForEach item in a list, create a new child, ask the child a question and add the result to a second list... and do it all async 'cause the questions are complicated.

v1rusw0rm

@v1rusw0rm

If I Ask instead of Tell (as he uses Tell in the example) I can't realistically bundle it into a task list, can I?

It's possible. Something like this:

var tasks = new List<Task<int>>();
foreach(var filename in filenames)
{
  var actor = CreateChild();
  tasks.Add(actor.Ask(filename, TimeSpan.FromSeconds(10)));
}
int[] lineCounts = await Task.WhenAll(tasks.ToArray());

But in your case I think it's better to use Tell instead of Ask. Do you create childs from actor or from outside?

matneyx

@matneyx

From inside their shared parent... 1 parent creates multiple children

Tell can't return a result though, can it?

I think your code sample might work. :)

v1rusw0rm

@v1rusw0rm

I wrote it without testing, so adapt it to your needs. The key here is a Task.WhenAll. Documentation: https://msdn.microsoft.com/en-us/library/hh194874(v=vs.110).aspx

Tell can't return a result though, can it?

It can't. But child can send result using Tell to parent.

matneyx

@matneyx

And then you have a separate receive in the parent for the message coming back from the child?

v1rusw0rm

@v1rusw0rm

Exactly.

matneyx

@matneyx

Awesome, thank you

And, considering my usage, I think you're right... Tell is going to be a better option.

v1rusw0rm

@v1rusw0rm

Tell is always better in performance-wise cases.

Good article, explaining when to use Ask: https://bartoszsypytkowski.com/dont-ask-tell-2/

Shukhrat Nekbaev

@snekbaev

Hi, question about .ConfigureAwait(false):
in ctor: ReceiveAsync<RequestAllInstances>( HandleRequestAllInstancesAsync );
method:

private async Task HandleRequestAllInstancesAsync( RequestAllInstances msg )
{
    var sender = Sender;

    var list = new List<A> { new A(), new A(), new A() };

    await Task.Delay( 10000 ).ConfigureAwait( false );

    // Sender.Tell( new RespondAllInstances( list.AsReadOnly() ) ); <-- this is going to crash with missing ActorContext
    sender.Tell( new RespondAllInstances( list.AsReadOnly() ) ); // but if saved beforehand then it's ok
}

actor system host app: regular console application

What is the general recommendation regarding the awaited tasks and captured context? Should I keep doing as in example above, i.e. capture context specific variables beforehand or should I switch to .ConfigureAwait(true)?

Sam13

@Sam13

@Danthar Thanks for your help. I ended up with something like that:

public FileManager()
{    
    m_fileTransferQueue = Source.Queue<Tuple<FileTransferStart, IActorRef>>(25, OverflowStrategy.Backpressure).To(Sink.ForEach<Tuple<FileTransferStart, IActorRef>>(DoProcess)).Run(Context.Materializer());
}

protected override void PostStop()
{
    base.PostStop();
    m_fileTransferQueue.Complete();
}

private void OnReceive(FileTransferStart fileTransferStart)
{
    m_fileTransferQueue.OfferAsync(new Tuple<FileTransferStart, IActorRef>(fileTransferStart, Sender));    
}

private void DoProcess(Tuple<FileTransferStart, IActorRef> fileTransferStart)
{
    var fileInfo = new FileInfo(Path.Combine(m_directory, fileTransferStart.Item1.FileName));
    var transferTask = FileIO.FromFile(fileInfo).Via(Flow.Create<ByteString>().Select(x => x.ToArray())).To(Sink.ActorRef<byte[]>(fileTransferStart.Item2, new FileTransferFinished())).Run(Context.Materializer()).ContinueWith(x =>
    {
        if (!x.Result.WasSuccessful)
        {
            Logger.ErrorException("Error during file transfer", x.Result.Error);
        }
    });
    // Run synchronous to have only 25 active items in queue?
    transferTask.Wait();
}

Is it correct that the processing of the file must run synchronous to ensure only 25 items are processed at the same time?

Arjen Smits

@Danthar

What the ForEach sink does. is it takes an item from the stream, runs its action, and takes the next. If your action is async, it will not block, and thus keep taking the next item. So yes, in this configuration. you would have to run your action syncronous.

Arjen Smits

@Danthar

@Sam13 take a look at the SelectAsync operator

And for more context and samples: https://getakka.net/articles/streams/integration.html

Sam13

@Sam13

@Danthar Ok thanks, just want to make sure I understood everything correctly. I'll have a look at that!

Vasily Kirichenko

@vasily-kirichenko

@Sam13 why do you not use struct tuples?

Bruno Alencar

@bruno-alencar

Can anyone help me?

I'm looking for a way to combine Rest API with Akka.net in a Cluster System. I would not connect the rest api receiver inside the cluster,
but sending messages for the cluster group.
I saw something about recepcionist, but when I tried to use inside the lighthouse system, It occurred exception because it was required that
know all messages types.

Is there any way to use these things together with best performance?

Sam13

@Sam13

@vasily-kirichenko I'm still on .NET 4.6.1, ValueTuple was introduced in 4.7

AndreSteenbergen

@AndreSteenbergen

Guys, what do you think? I created an Unbounded ---Stable--- PriorityQueue. Is this Gist worthy of creating a Pull Request to fix: akkadotnet/akka.net#2652

https://gist.github.com/AndreSteenbergen/a2260f23c4e7d2505364f2271440d30d

nathvi

@nathvi

@AndreSteenbergen , I would change Seqnum to SequenceNumber

If you're going to name that Func priorityCalculator instead of priorityCalc, might as well be equally verbose.

nathvi

@nathvi

Also I'd change the line:
var ci = _data.Count - 1; // child index; start at end
To something like:
var endChildIndex = _data.Count - 1;

@AndreSteenbergen

AndreSteenbergen

@AndreSteenbergen

I am actually not so much inclined to change that code, as it is a straight copy from the ListPriorityQueue from akka.net sources on github

seqnum and priorityCalc yes, I think you are correct.

nathvi

@nathvi

If you have to add a comment to explain what the intention of the statement is, then the variable names are poor.

Not always, but in this case, I think having a more explicit name for the var would be much better than littering the code with comments.

Where communities thrive

People

Repo info

Activity