/
benchmarks.ex
140 lines (116 loc) · 3.8 KB
/
benchmarks.ex
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
defmodule ElixirBench.Benchmarks do
import Ecto
import Ecto.Query, warn: false
alias ElixirBench.Repo
alias Ecto.Multi
alias ElixirBench.Github
alias ElixirBench.Benchmarks.{Benchmark, Measurement, Job, Runner, Config}
def data() do
Dataloader.Ecto.new(Repo, query: &query/2)
end
def query(Benchmark, %{repo_id: repo_id}) do
from(b in Benchmark, where: b.repo_id == ^repo_id)
end
def query(queryable, _args) do
queryable
end
def create_runner(attrs) do
%Runner{}
|> Runner.changeset(attrs)
|> Repo.insert()
end
def authenticate_runner(name, api_key) do
with {:ok, runner} <- Repo.fetch(where(Runner, name: ^name)) do
if Runner.verify_api_key?(runner, api_key) do
{:ok, runner}
else
{:error, :not_found}
end
end
end
def fetch_benchmark(repo_id, name) do
Repo.fetch(where(Benchmark, repo_id: ^repo_id, name: ^name))
end
def fetch_measurement(id) do
Repo.fetch(where(Measurement, id: ^id))
end
def fetch_job(id) do
Repo.fetch(where(Job, id: ^id))
end
def fetch_job_by_uuid(uuid) do
Repo.fetch(where(Job, uuid: ^uuid))
end
def list_benchmarks_by_repo_id(repo_ids) do
Repo.all(from(b in Benchmark, where: b.repo_id in ^repo_ids))
end
def list_jobs_by_repo_id(repo_ids) do
Repo.all(from(j in Job, where: j.repo_id in ^repo_ids))
end
def list_jobs() do
Repo.all(Job)
end
def create_job(repo, attrs) do
changeset = Job.create_changeset(%Job{repo_id: repo.id}, attrs)
with {:ok, job} <- Ecto.Changeset.apply_action(changeset, :insert),
{:ok, raw_config} <- Github.fetch_config(repo.owner, repo.name, job.commit_sha) do
config_changeset = Config.changeset(%Config{}, raw_config)
changeset
|> Ecto.Changeset.put_embed(:config, config_changeset)
|> Repo.insert()
end
end
def claim_job(%Runner{} = runner) do
Repo.transaction(fn ->
with {:ok, job} <- fetch_unclaimed_job(runner) do
changeset = Job.claim_changeset(job, runner.id)
Repo.update!(changeset)
else
{:error, error} -> Repo.rollback(error)
end
end)
end
def submit_job(%Job{} = job, results) do
multi = Multi.update(Multi.new(), :job, Job.submit_changeset(job, results))
multi =
Enum.reduce(results["measurements"] || [], multi, fn {name, result}, multi ->
benchmark = {:benchmark, name}
multi
|> Multi.run(benchmark, fn _ ->
{:ok, get_or_create_benchmark!(job.repo_id, name)}
end)
|> Multi.run({:measurement, name}, fn %{^benchmark => benchmark} ->
create_measurement(benchmark, job, result)
end)
end)
case Repo.transaction(multi) do
{:ok, _} -> :ok
{:error, _, changeset, _} -> {:error, changeset}
end
end
defp get_or_create_benchmark!(repo_id, name) do
# We need to set something on_conflict, otherwise, we won't be able to get the id
# of already existing result in returning
updates = [updated_at: DateTime.utc_now()]
opts = [on_conflict: [set: updates], conflict_target: [:repo_id, :name], returning: true]
Repo.insert!(%Benchmark{repo_id: repo_id, name: name}, opts)
end
defp create_measurement(%Benchmark{} = bench, %Job{} = job, attrs) do
build_assoc(bench, :measurements, job_id: job.id)
|> Measurement.changeset(attrs)
|> Repo.insert()
end
defp fetch_unclaimed_job(runner) do
# Unclaimed or claimed by this runner but not completed
Repo.fetch(
from(
j in Job,
where: is_nil(j.claimed_by) and is_nil(j.claimed_at) and is_nil(j.completed_at),
or_where:
j.claimed_by == ^runner.id and not is_nil(j.claimed_at) and is_nil(j.completed_at),
lock: "FOR UPDATE SKIP LOCKED",
order_by: j.inserted_at,
limit: 1
)
)
end
end