- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在 Elixir 中建立一个工作队列作为学术练习。目前,我的工作人员必须在创建队列时手动注册自己(参见 MyQuestion.Worker.start_link
)。
我希望我的主管在创建/重新启动时将可用的工作人员注册到队列中,因为这似乎有助于测试工作人员并最大限度地减少耦合。
有没有办法做我在 MyQuestion.Supervisor
中的代码中描述的事情?
defmodule MyQuestion.Supervisor do
use Supervisor
def start_link do
supervisor = Supervisor.start_link(__MODULE__, :ok)
end
def init(:ok) do
children = [
worker(MyQuestion.JobQueue, []),
worker(MyQuestion.Worker, [], id: :worker_0),
worker(MyQuestion.Worker, [], id: :worker_1)]
supervise(children, strategy: :rest_for_one)
end
# LOOKING FOR SOMETHING LIKE THIS
# on worker spawn, I want to add the worker to the queue
def child_spawned(pid, {MyQuestion.Worker, _, _}) do
# add worker to queue
MyQuestion.JobQueue.add_new_worker(pid)
end
# LOOKING FOR SOMETHING LIKE THIS
# I want some way to do the following (imagine the callback existed)
def child_terminated(pid, reason, state)
# with this information I could tell the job queue to mark
# the job associated with the pid as failed and to retry
# or maybe extract the job id from the worker state, etc.
MyQuestion.JobQueue.remove_worker(pid)
MyQuestion.JobQueue.restart_job_for_failed_worker(pid)
end
end
defmodule MyQuestion.JobQueue do
def start_link do
Agent.start_link(fn -> [] end, name: __MODULE__)
end
def new_worker(pid) do
# register pid with agent state in available worker list, etc.
end
def add_job(job_description) do
# find idle worker and run job
<... snip ...>
end
<... snip ...>
end
defmodule MyQuestion.Worker do
use GenServer
def start_link do
# start worker
{:ok, worker} = GenServer.start_link(__MODULE__, [])
# Now we have a worker pid, so we can register that pid with the queue
# I wish this could be in the supervisor or else where.
MyQuestion.JobQueue.add_new_worker(worker)
# must return gen server's start link
{:ok, worker}
end
<... snip ...>
end
最佳答案
他们的关键是调用 Process.monitor(pid)
的组合。 – 然后您将接到 handle_info
的电话– 并手动调用 Supervisor.start_child
这给了你pid。
我之前曾尝试使用 handle_info
但永远无法调用它。 Process.monitor(pid)
必须从您希望接收通知的同一进程中调用,因此您必须从 handle_call
内部调用它函数将监视器与您的服务器进程相关联。可能有一个函数可以将代码作为另一个进程运行(即 run_from_process(job_queue_pid, fn -> Process.monitor(pid_to_monitor) end)
),但我找不到任何东西。
附件是一个非常幼稚的作业队列实现。我只在 Elixir 呆了一天,所以代码既困惑又不惯用,但我附上它是因为似乎缺乏围绕该主题的示例代码。
看HeavyIndustry.JobQueue
, handle_info
, create_new_worker
.这段代码有一个明显的问题:它能够在工作人员崩溃时重新启动它们,但它无法从该代码启动下一个作业的队列(由于需要 GenServer.call
内的 handle_info
,这使我们陷入僵局)。我认为您可以通过将启 Action 业的进程与跟踪作业的进程分开来解决此问题。如果您运行示例代码,您会注意到它最终会停止运行作业,即使队列中还有一个作业(:crash
作业)。
defmodule HeavyIndustry.Supervisor do
use Supervisor
def start_link do
Supervisor.start_link(__MODULE__, :ok)
end
def init(:ok) do
# default to supervising nothing, we will add
supervise([], strategy: :one_for_one)
end
def create_children(supervisor, worker_count) do
# create the job queue. defaults to no workers
Supervisor.start_child(supervisor, worker(HeavyIndustry.JobQueue, [[supervisor, worker_count]]))
end
end
defmodule HeavyIndustry.JobQueue do
use GenServer
@job_queue_name __MODULE__
def start_link(args, _) do
GenServer.start_link(__MODULE__, args, name: @job_queue_name)
end
def init([supervisor, n]) do
# set some default state
state = %{
supervisor: supervisor,
max_workers: n,
jobs: [],
workers: %{
idle: [],
busy: []
}
}
{:ok, state}
end
def setup() do
# we want to be aware of worker failures. we hook into this by calling
# Process.monitor(pid), but this links the calling process with the monitored
# process. To make sure the calls come to US and not the process that called
# setup, we create the workers by passing a message to our server process
state = GenServer.call(@job_queue_name, :setup)
# gross passing the whole state back here to monitor but the monitoring must
# be started from the server process and we can't call GenServer.call from
# inside the :setup call else we deadlock.
workers = state.workers.idle
GenServer.call(@job_queue_name, {:monitor_pids, workers})
end
def add_job(from, job) do
# add job to queue
{:ok, our_job_id} = GenServer.call(@job_queue_name, {:create_job, %{job: job, reply_to: from}})
# try to run the next job
case GenServer.call(@job_queue_name, :start_next_job) do
# started our job
{:ok, started_job_id = ^our_job_id} -> {:ok, :started}
# started *a* job
{:ok, _} -> {:ok, :pending}
# couldnt start any job but its ok...
{:error, :no_idle_workers} -> {:ok, :pending}
# something fell over...
{:error, e} -> {:error, e}
# yeah I know this is bad.
_ -> {:ok}
end
end
def start_next_job do
GenServer.call(@job_queue_name, :start_next_job)
end
##
# Internal API
##
def handle_call(:setup, _, state) do
workers = Enum.map(0..(state.max_workers-1), fn (n) ->
{:ok, pid} = start_new_worker(state.supervisor)
pid
end)
state = %{state | workers: %{state.workers | idle: workers}}
{:reply, state, state}
end
defp start_new_worker(supervisor) do
spec = Supervisor.Spec.worker(HeavyIndustry.Worker, [], id: :"Worker.#{:os.system_time}", restart: :temporary)
# start worker
Supervisor.start_child(supervisor, spec)
end
def handle_call({:monitor_pids, list}, _, state) do
Enum.each(list, &Process.monitor(&1))
{:reply, :ok, state}
end
def handle_call({:create_job, job}, from, state) do
job = %{
job: job.job,
reply_to: job.reply_to,
id: :os.system_time, # id for task
status: :pending, # start pending, go active, then remove
pid: nil
}
# add new job to jobs list
state = %{state | jobs: state.jobs ++ [job]}
{:reply, {:ok, job.id}, state}
end
def handle_call(:start_next_job, _, state) do
IO.puts "==> Start Next Job"
IO.inspect state
IO.puts "=================="
reply = case {find_idle_worker(state.workers), find_next_job(state.jobs)} do
{{:error, :no_idle_workers}, _} ->
# no workers for job, doesnt matter if we have a job
{:error, :no_idle_workers}
{_, nil} ->
# no job, doesnt matter if we have a worker
{:error, :no_more_jobs}
{{:ok, worker}, job} ->
# have worker, have job, do work
# update state to set job active and worker busy
jobs = state.jobs -- [job]
job = %{job | status: :active, pid: worker}
jobs = jobs ++ [job]
idle = state.workers.idle -- [worker]
busy = state.workers.busy ++ [worker]
state = %{state | jobs: jobs, workers: %{idle: idle, busy: busy}}
{:ok, task_id} = Task.start(fn ->
result = GenServer.call(worker, job.job)
remove_job(job)
free_worker(worker)
send job.reply_to, %{answer: result, job: job.job}
start_next_job
end)
{:ok, job.id}
end
{:reply, reply, state}
end
defp find_idle_worker(workers) do
case workers do
%{idle: [], busy: _} -> {:error, :no_idle_workers}
%{idle: [worker | idle], busy: busy} -> {:ok, worker}
end
end
defp find_next_job(jobs) do
jobs |> Enum.find(&(&1.status == :pending))
end
defp free_worker(worker) do
GenServer.call(@job_queue_name, {:free_worker, worker})
end
defp remove_job(job) do
GenServer.call(@job_queue_name, {:remove_job, job})
end
def handle_call({:free_worker, worker}, from, state) do
idle = state.workers.idle ++ [worker]
busy = state.workers.busy -- [worker]
{:reply, :ok, %{state | workers: %{idle: idle, busy: busy}}}
end
def handle_call({:remove_job, job}, from, state) do
jobs = state.jobs -- [job]
{:reply, :ok, %{state | jobs: jobs}}
end
def handle_info(msg = {reason, ref, :process, pid, _reason}, state) do
IO.puts "Worker collapsed: #{reason} #{inspect pid}, clear and restart job"
# find job for collapsed worker
# set job to pending again
job = Enum.find(state.jobs, &(&1.pid == pid))
fixed_job = %{job | status: :pending, pid: nil}
jobs = (state.jobs -- [job]) ++ [fixed_job]
# remote worker from lists
idle = state.workers.idle -- [pid]
busy = state.workers.busy -- [pid]
# start new worker
{:ok, pid} = start_new_worker(state.supervisor)
# add worker from lists
idle = state.workers.idle ++ [pid]
# cant call GenServer.call from here to monitor pid,
# so duplicate the code a bit...
Process.monitor(pid)
# update state
state = %{state | jobs: jobs, workers: %{idle: idle, busy: busy}}
{:noreply, state}
end
end
defmodule HeavyIndustry.Worker do
use GenServer
def start_link do
GenServer.start_link(__MODULE__, :ok)
end
def init(:ok) do
# workers have no persistent state
IO.puts "==> Worker up! #{inspect self}"
{:ok, nil}
end
def handle_call({:sum, list}, from, _) do
sum = Enum.reduce(list, fn (n, acc) -> acc + n end)
{:reply, sum, nil}
end
def handle_call({:fib, n}, from, _) do
sum = fib_calc(n)
{:reply, sum, nil}
end
def handle_call({:stop}, from, state) do
{:stop, "my-stop-reason", "my-stop-reply", state}
end
def handle_call({:crash}, from, _) do
{:reply, "this will crash" ++ 1234, nil}
end
def handle_call({:timeout}, from, _) do
:timer.sleep 10000
{:reply, "this will timeout", nil}
end
# Slow fib
defp fib_calc(0), do: 0
defp fib_calc(1), do: 1
defp fib_calc(n), do: fib_calc(n-1) + fib_calc(n-2)
end
defmodule Looper do
def start do
{:ok, pid} = HeavyIndustry.Supervisor.start_link
{:ok, job_queue} = HeavyIndustry.Supervisor.create_children(pid, 2)
HeavyIndustry.JobQueue.setup()
add_jobs
loop
end
def add_jobs do
jobs = [
{:sum, [100, 200, 300]},
{:crash},
{:fib, 35},
{:fib, 35},
{:sum, [88, 88, 99]},
{:fib, 35},
{:fib, 35},
{:fib, 35},
{:sum, 0..100},
# {:stop}, # stop not really a failure
{:sum, [88, 88, 99]},
# {:timeout},
{:sum, [-1]}
]
Enum.each(jobs, fn (job) ->
IO.puts "~~~~> Add job: #{inspect job}"
case HeavyIndustry.JobQueue.add_job(self, job) do
{:ok, :started} -> IO.puts "~~~~> Started job immediately"
{:ok, :pending} -> IO.puts "~~~~> Job in queue"
val -> IO.puts "~~~~> ... val: #{inspect val}"
end
end)
end
def loop do
receive do
value ->
IO.puts "~~~~> Received: #{inspect value}"
loop
end
end
end
Looper.start
关于elixir - 检测 Elixir/OTP 主 pipe 生成和终止,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/33187041/
如果我终止应用程序,我在尝试保持我的功能运行时卡住了。 是否可以在应用程序未运行时保持核心位置(地理围栏/地理定位)和核心蓝牙运行?如果可能如何解决我的问题?我已经检查了背景模式,并实现了核心定位方法
该程序要求用户输入一个数字,然后从列表中返回详细信息。我该怎么做? do { Scanner in = new Scanner(System.in);
我正在开发一个内部分发的 iOS 应用程序(即,没有应用程序商店),我希望能够以恒定的 10 分钟间隔报告设备的位置。 无论如何,我在我的 plist 中包含了 location 作为字段 UIBac
我的 mongodb 服务器突然收到信号 15(终止)。我不知道为什么 mongodb 崩溃了。以下是日志消息。 Mon Jun 27 07:33:31.701 [signalProcessingTh
我按顺序运行了一堆malloc,并且每次都检查以确保它是成功的。像这样: typedef struct { int *aray; char *string; } mystruct; m
这个问题已经有答案了: How to stop a running pthread thread? (4 个回答) 已关闭 8 年前。 可以使用 pthread_join() 停止线程。但让我们想象一
#include #include #include struct node{ char data; int p; struct node *ptr; }; struct node *st
这个问题已经有答案了: Why should I use a semicolon after every function in javascript? (9 个回答) 已关闭 8 年前。 好吧,我问
我有一个启动多个工作线程的函数。每个工作线程都由一个对象封装,该对象的析构函数将尝试加入线程,即调用if (thrd_.joinable()) thrd_.join();。但是,每个 worker 必
我正在实现一个应用程序,当用户摇动手机时,该应用程序会监听并采取行动。 所以我实现了以下服务: public class ShakeMonitorService extends Service {
我在使用 Xcode 时遇到问题,其中弹出错误“Source Kit Service Terminated”,并且所有语法突出显示和代码完成在 Swift 中都消失了。我怎样才能解决这个问题? 这是一
我想为我的控制台应用程序安全退出,该应用程序将使用单声道在 linux 上运行,但我找不到解决方案来检测信号是否发送到它或用户是否按下了 ctrl+c。 在 Windows 上有内核函数 SetCon
关键: pthread_cancel函数发送终止信号pthread_setcancelstate函数设置终止方式pthread_testcancel函数取消线程(另一功能是:设置取消点) 1 线程取消
下面的程序在不同的选项级别下有不同的行为。当我用 -O3 编译它时,它永远不会终止。当我用 -O0 编译它时,它总是很快就会终止。 #include #include void *f(void *
我有 3 个节点的 K8S 集群,我创建了 3 个副本 pod,应用程序 app1 在所有 pod 上运行,我通过运行 service yaml 文件建立了服务,我可以看到通过运行 kubectl g
我打算使用 nginx 来代理 websocket。在执行 nginx reload/HUP 时,我知道 nginx 等待旧的工作进程停止处理所有请求。然而,在 websocket 连接中,这可能不会
在 Ubuntu 9.10 上使用 PVM 3.4.5-12(使用 apt-get 时的 PVM 包) 添加主机后程序终止。 laptop> pvm pvm> add bowtie-slave add
我编写了一个应用程序来从 iPhone 录制视频。它工作正常,但有一个大问题。当 AVCaptureSession 开始运行并且用户尝试从其库(iPod)播放音频时。此操作将使 AVCaptureSe
我将如何使用NSRunningApplication?我有与启动应用程序相反的东西: [[NSWorkspace sharedWorkspace] launchApplication:appName]
我正在使用 NSTask 执行一系列长时间运行的命令,如下所示: commandToRun = @"command 1;command2"; NSArray *arguments = [NSArray
我是一名优秀的程序员,十分优秀!