преобразование блокировки потока для блокировки без потока в f #

Question

Я извлекаю данные из API Bloomberg и удивляюсь медленности. Мое вычисление ограничено этим.

Поэтому я решил использовать какой-то асинхронный построитель монады, чтобы развязать его. После запуска результаты не намного лучше, что было очевидно, когда я делаю вызов функции NextEvent, которая является блокировкой потоков.

 let outerloop args dic = 
     ... 
     let rec innerloop continuetoloop = 
      let eventObj = session.NextEvent(); //This blocks 
      ... 

    let seqtable = reader.ReadFile(@"C:\homeware\sector.csv", ";".[0], true) 

    let dic = ConcurrentDictionary<_,_>() 
    let wf = seqtable |> Seq.mapi (fun i item -> async { outerloop item dic }) 
    wf |> Async.Parallel 
     |> Async.RunSynchronously 
     |> ignore 
    printfn "%A" ret 

Есть хороший способ, чтобы обернуть, что блокирующий вызов к блокирующим вызову? Кроме того, почему структура async не создает столько потоков, сколько у меня есть запросы (например, 200)? когда я инспектировать нити, из которых я получаю значение я вижу только 4-5, которые используются ..

UPDATE

Я нашел убедительную причину, почему он никогда не будет возможно. Операция async берет то, что после инструкции async, и планируйте ее где-то в threadpool. для всего, что имеет значение, до тех пор, пока функция асинхронизации используется правильно, то есть, всегда возвращаясь к файлу threadpool, от которого он произошел, мы можем считать, что мы выполняемся в одном потоке.

Будучи на одном потоке, все планирование должно выполняться где-то позже, а команда блокировки не может избежать того факта, что, в конце концов, после ее запуска ему придется блокировать в какой-то момент в будущем поток.

Answer 1

Есть ли хороший способ обернуть этот блокирующий вызов неблокируемому звонку?

Нет. Вы никогда не сможете переносить блокирующие вызовы, чтобы сделать их неблокирующими. Если бы вы могли, асинк был бы просто шаблоном проектирования, а не фундаментальным сдвигом парадигмы.

Кроме того, почему инфраструктура async не создает столько потоков, сколько у меня есть запросы (например, 200)?

Async построен на пуле потоков, который разработан, чтобы не создавать потоки агрессивно, потому что они настолько дороги. Весь смысл (реального) async состоит в том, что он не требует столько потоков, сколько есть соединений. Вы можете обрабатывать 10 000 одновременных соединений с ~ 30 потоками.

Возможно, у вас есть полное непонимание того, что такое асинхронное и для чего оно предназначено. Я бы предложил купить любую из книг F #, которые охватывают эту тему и читают ее. В частности, ваше решение не является асинхронным, потому что вы просто вызываете свой блокирующий элемент StartGetFieldsValue из вашего рабочего процесса async, победив цель async. Вместо этого вы можете просто сделать Array.Parallel.map getFieldsValue.

Кроме того, вы хотите использовать чисто функциональный API, когда делаете что-то параллельно, а не мутируете ConcurrentDictionary на месте. Так заменить req.StartGetFieldsValue ret с

let! result = req.StartGetFieldsValue() 
... 
return result 

и заменить ignore с dict.

Answer 2

Вот решение, которое я сделал, что, похоже, работает. Конечно, он не использует только async (минус a), но также Async.

Я определяю простой тип, который имеет одно событие, законченное и один метод asyncstart, с помощью метода для запуска в качестве аргумента. он запускает метод, затем срабатывает событие закончилось в соответствующем месте (в моем случае я должен был захватить контекст синхронизации и т.д ..)

Тогда в стороне потребителя, я использую просто

let! completion = Async.Waitfromevent wapper.finished |> Async.StartAsChild 
let! completed = completion 

В то время как запуск этого кода, со стороны потребителя, я использую только асинхронные вызовы, что делает мой код незаблокированным. Конечно, где-то есть какой-то поток, который заблокирован, но это происходит за пределами моей основной рабочей петли, которая остается реактивной и подходящей.