git.squeep.com Git - akkoma/blob - benchmarks/load_testing/activities.ex

   1 defmodule Pleroma.LoadTesting.Activities do
   2   @moduledoc """
   3   Module for generating different activities.
   4   """
   5   import Ecto.Query
   6   import Pleroma.LoadTesting.Helper, only: [to_sec: 1]
   7
   8   alias Ecto.UUID
   9   alias Pleroma.Constants
  10   alias Pleroma.LoadTesting.Users
  11   alias Pleroma.Repo
  12   alias Pleroma.Web.CommonAPI
  13
  14   require Constants
  15
  16   @defaults [
  17     iterations: 170,
  18     friends_used: 20,
  19     non_friends_used: 20
  20   ]
  21
  22   @max_concurrency 10
  23
  24   @visibility ~w(public private direct unlisted)
  25   @types [
  26     :simple,
  27     :emoji,
  28     :mentions,
  29     :hell_thread,
  30     :attachment,
  31     :tag,
  32     :like,
  33     :reblog,
  34     :simple_thread
  35   ]
  36   @groups [:friends_local, :friends_remote, :non_friends_local, :non_friends_local]
  37   @remote_groups [:friends_remote, :non_friends_remote]
  38   @friends_groups [:friends_local, :friends_remote]
  39   @non_friends_groups [:non_friends_local, :non_friends_remote]
  40
  41   @spec generate(User.t(), keyword()) :: :ok
  42   def generate(user, opts \\ []) do
  43     {:ok, _} =
  44       Agent.start_link(fn -> %{} end,
  45         name: :benchmark_state
  46       )
  47
  48     opts = Keyword.merge(@defaults, opts)
  49
  50     users = Users.prepare_users(user, opts)
  51
  52     {:ok, _} = Agent.start_link(fn -> users[:non_friends_remote] end, name: :non_friends_remote)
  53
  54     task_data =
  55       for visibility <- @visibility,
  56           type <- @types,
  57           group <- [:user | @groups],
  58           do: {visibility, type, group}
  59
  60     IO.puts("Starting generating #{opts[:iterations]} iterations of activities...")
  61
  62     public_long_thread = fn ->
  63       generate_long_thread("public", users, opts)
  64     end
  65
  66     private_long_thread = fn ->
  67       generate_long_thread("private", users, opts)
  68     end
  69
  70     iterations = opts[:iterations]
  71
  72     {time, _} =
  73       :timer.tc(fn ->
  74         Enum.each(
  75           1..iterations,
  76           fn
  77             i when i == iterations - 2 ->
  78               spawn(public_long_thread)
  79               spawn(private_long_thread)
  80               generate_activities(users, Enum.shuffle(task_data), opts)
  81
  82             _ ->
  83               generate_activities(users, Enum.shuffle(task_data), opts)
  84           end
  85         )
  86       end)
  87
  88     IO.puts("Generating iterations of activities took #{to_sec(time)} sec.\n")
  89     :ok
  90   end
  91
  92   def generate_power_intervals(opts \\ []) do
  93     count = Keyword.get(opts, :count, 20)
  94     power = Keyword.get(opts, :power, 2)
  95     IO.puts("Generating #{count} intervals for a power #{power} series...")
  96     counts = Enum.map(1..count, fn n -> :math.pow(n, power) end)
  97     sum = Enum.sum(counts)
  98
  99     densities =
 100       Enum.map(counts, fn c ->
 101         c / sum
 102       end)
 103
 104     densities
 105     |> Enum.reduce(0, fn density, acc ->
 106       if acc == 0 do
 107         [{0, density}]
 108       else
 109         [{_, lower} | _] = acc
 110         [{lower, lower + density} | acc]
 111       end
 112     end)
 113     |> Enum.reverse()
 114   end
 115
 116   def generate_tagged_activities(opts \\ []) do
 117     tag_count = Keyword.get(opts, :tag_count, 20)
 118     users = Keyword.get(opts, :users, Repo.all(Pleroma.User))
 119     activity_count = Keyword.get(opts, :count, 200_000)
 120
 121     intervals = generate_power_intervals(count: tag_count)
 122
 123     IO.puts(
 124       "Generating #{activity_count} activities using #{tag_count} different tags of format `tag_n`, starting at tag_0"
 125     )
 126
 127     Enum.each(1..activity_count, fn _ ->
 128       random = :rand.uniform()
 129       i = Enum.find_index(intervals, fn {lower, upper} -> lower <= random && upper > random end)
 130       CommonAPI.post(Enum.random(users), %{status: "a post with the tag #tag_#{i}"})
 131     end)
 132   end
 133
 134   defp generate_long_thread(visibility, users, _opts) do
 135     group =
 136       if visibility == "public",
 137         do: :friends_local,
 138         else: :user
 139
 140     tasks = get_reply_tasks(visibility, group) |> Stream.cycle() |> Enum.take(50)
 141
 142     {:ok, activity} =
 143       CommonAPI.post(users[:user], %{
 144         status: "Start of #{visibility} long thread",
 145         visibility: visibility
 146       })
 147
 148     Agent.update(:benchmark_state, fn state ->
 149       key =
 150         if visibility == "public",
 151           do: :public_thread,
 152           else: :private_thread
 153
 154       Map.put(state, key, activity)
 155     end)
 156
 157     acc = {activity.id, ["@" <> users[:user].nickname, "reply to long thread"]}
 158     insert_replies_for_long_thread(tasks, visibility, users, acc)
 159     IO.puts("Generating #{visibility} long thread ended\n")
 160   end
 161
 162   defp insert_replies_for_long_thread(tasks, visibility, users, acc) do
 163     Enum.reduce(tasks, acc, fn
 164       :user, {id, data} ->
 165         user = users[:user]
 166         insert_reply(user, List.delete(data, "@" <> user.nickname), id, visibility)
 167
 168       group, {id, data} ->
 169         replier = Enum.random(users[group])
 170         insert_reply(replier, List.delete(data, "@" <> replier.nickname), id, visibility)
 171     end)
 172   end
 173
 174   defp generate_activities(users, task_data, opts) do
 175     Task.async_stream(
 176       task_data,
 177       fn {visibility, type, group} ->
 178         insert_activity(type, visibility, group, users, opts)
 179       end,
 180       max_concurrency: @max_concurrency,
 181       timeout: 30_000
 182     )
 183     |> Stream.run()
 184   end
 185
 186   defp insert_local_activity(visibility, group, users, status) do
 187     {:ok, _} =
 188       group
 189       |> get_actor(users)
 190       |> CommonAPI.post(%{status: status, visibility: visibility})
 191   end
 192
 193   defp insert_remote_activity(visibility, group, users, status) do
 194     actor = get_actor(group, users)
 195     {act_data, obj_data} = prepare_activity_data(actor, visibility, users[:user])
 196     {activity_data, object_data} = other_data(actor, status)
 197
 198     activity_data
 199     |> Map.merge(act_data)
 200     |> Map.put("object", Map.merge(object_data, obj_data))
 201     |> Pleroma.Web.ActivityPub.ActivityPub.insert(false)
 202   end
 203
 204   defp user_mentions(users) do
 205     user_mentions =
 206       Enum.reduce(
 207         @groups,
 208         [],
 209         fn group, acc ->
 210           acc ++ get_random_mentions(users[group], Enum.random(0..2))
 211         end
 212       )
 213
 214     if Enum.random([true, false]),
 215       do: ["@" <> users[:user].nickname | user_mentions],
 216       else: user_mentions
 217   end
 218
 219   defp hell_thread_mentions(users) do
 220     with {:ok, nil} <- Cachex.get(:user_cache, "hell_thread_mentions") do
 221       cached =
 222         @groups
 223         |> Enum.reduce([users[:user]], fn group, acc ->
 224           acc ++ Enum.take(users[group], 5)
 225         end)
 226         |> Enum.map(&"@#{&1.nickname}")
 227         |> Enum.join(", ")
 228
 229       Cachex.put(:user_cache, "hell_thread_mentions", cached)
 230       cached
 231     else
 232       {:ok, cached} -> cached
 233     end
 234   end
 235
 236   defp insert_activity(:simple, visibility, group, users, _opts)
 237        when group in @remote_groups do
 238     insert_remote_activity(visibility, group, users, "Remote status")
 239   end
 240
 241   defp insert_activity(:simple, visibility, group, users, _opts) do
 242     insert_local_activity(visibility, group, users, "Simple status")
 243   end
 244
 245   defp insert_activity(:emoji, visibility, group, users, _opts)
 246        when group in @remote_groups do
 247     insert_remote_activity(visibility, group, users, "Remote status with emoji :firefox:")
 248   end
 249
 250   defp insert_activity(:emoji, visibility, group, users, _opts) do
 251     insert_local_activity(visibility, group, users, "Simple status with emoji :firefox:")
 252   end
 253
 254   defp insert_activity(:mentions, visibility, group, users, _opts)
 255        when group in @remote_groups do
 256     mentions = user_mentions(users)
 257
 258     status = Enum.join(mentions, ", ") <> " remote status with mentions"
 259
 260     insert_remote_activity(visibility, group, users, status)
 261   end
 262
 263   defp insert_activity(:mentions, visibility, group, users, _opts) do
 264     mentions = user_mentions(users)
 265
 266     status = Enum.join(mentions, ", ") <> " simple status with mentions"
 267     insert_remote_activity(visibility, group, users, status)
 268   end
 269
 270   defp insert_activity(:hell_thread, visibility, group, users, _)
 271        when group in @remote_groups do
 272     mentions = hell_thread_mentions(users)
 273     insert_remote_activity(visibility, group, users, mentions <> " remote hell thread status")
 274   end
 275
 276   defp insert_activity(:hell_thread, visibility, group, users, _opts) do
 277     mentions = hell_thread_mentions(users)
 278
 279     insert_local_activity(visibility, group, users, mentions <> " hell thread status")
 280   end
 281
 282   defp insert_activity(:attachment, visibility, group, users, _opts) do
 283     actor = get_actor(group, users)
 284
 285     obj_data = %{
 286       "actor" => actor.ap_id,
 287       "name" => "4467-11.jpg",
 288       "type" => "Document",
 289       "url" => [
 290         %{
 291           "href" =>
 292             "#{Pleroma.Web.base_url()}/media/b1b873552422a07bf53af01f3c231c841db4dfc42c35efde681abaf0f2a4eab7.jpg",
 293           "mediaType" => "image/jpeg",
 294           "type" => "Link"
 295         }
 296       ]
 297     }
 298
 299     object = Repo.insert!(%Pleroma.Object{data: obj_data})
 300
 301     {:ok, _activity} =
 302       CommonAPI.post(actor, %{
 303         status: "Post with attachment",
 304         visibility: visibility,
 305         media_ids: [object.id]
 306       })
 307   end
 308
 309   defp insert_activity(:tag, visibility, group, users, _opts) do
 310     insert_local_activity(visibility, group, users, "Status with #tag")
 311   end
 312
 313   defp insert_activity(:like, visibility, group, users, opts) do
 314     actor = get_actor(group, users)
 315
 316     with activity_id when not is_nil(activity_id) <- get_random_create_activity_id(),
 317          {:ok, _activity} <- CommonAPI.favorite(actor, activity_id) do
 318       :ok
 319     else
 320       {:error, _} ->
 321         insert_activity(:like, visibility, group, users, opts)
 322
 323       nil ->
 324         Process.sleep(15)
 325         insert_activity(:like, visibility, group, users, opts)
 326     end
 327   end
 328
 329   defp insert_activity(:reblog, visibility, group, users, opts) do
 330     actor = get_actor(group, users)
 331
 332     with activity_id when not is_nil(activity_id) <- get_random_create_activity_id(),
 333          {:ok, _activity} <- CommonAPI.repeat(activity_id, actor) do
 334       :ok
 335     else
 336       {:error, _} ->
 337         insert_activity(:reblog, visibility, group, users, opts)
 338
 339       nil ->
 340         Process.sleep(15)
 341         insert_activity(:reblog, visibility, group, users, opts)
 342     end
 343   end
 344
 345   defp insert_activity(:simple_thread, "direct", group, users, _opts) do
 346     actor = get_actor(group, users)
 347     tasks = get_reply_tasks("direct", group)
 348
 349     list =
 350       case group do
 351         :user ->
 352           group = Enum.random(@friends_groups)
 353           Enum.take(users[group], 3)
 354
 355         _ ->
 356           Enum.take(users[group], 3)
 357       end
 358
 359     data = Enum.map(list, &("@" <> &1.nickname))
 360
 361     {:ok, activity} =
 362       CommonAPI.post(actor, %{
 363         status: Enum.join(data, ", ") <> "simple status",
 364         visibility: "direct"
 365       })
 366
 367     acc = {activity.id, ["@" <> users[:user].nickname | data] ++ ["reply to status"]}
 368     insert_direct_replies(tasks, users[:user], list, acc)
 369   end
 370
 371   defp insert_activity(:simple_thread, visibility, group, users, _opts) do
 372     actor = get_actor(group, users)
 373     tasks = get_reply_tasks(visibility, group)
 374
 375     {:ok, activity} =
 376       CommonAPI.post(users[:user], %{status: "Simple status", visibility: visibility})
 377
 378     acc = {activity.id, ["@" <> actor.nickname, "reply to status"]}
 379     insert_replies(tasks, visibility, users, acc)
 380   end
 381
 382   defp get_actor(:user, %{user: user}), do: user
 383   defp get_actor(group, users), do: Enum.random(users[group])
 384
 385   defp other_data(actor, content) do
 386     %{host: host} = URI.parse(actor.ap_id)
 387     datetime = DateTime.utc_now()
 388     context_id = "https://#{host}/contexts/#{UUID.generate()}"
 389     activity_id = "https://#{host}/activities/#{UUID.generate()}"
 390     object_id = "https://#{host}/objects/#{UUID.generate()}"
 391
 392     activity_data = %{
 393       "actor" => actor.ap_id,
 394       "context" => context_id,
 395       "id" => activity_id,
 396       "published" => datetime,
 397       "type" => "Create",
 398       "directMessage" => false
 399     }
 400
 401     object_data = %{
 402       "actor" => actor.ap_id,
 403       "attachment" => [],
 404       "attributedTo" => actor.ap_id,
 405       "bcc" => [],
 406       "bto" => [],
 407       "content" => content,
 408       "context" => context_id,
 409       "conversation" => context_id,
 410       "emoji" => %{},
 411       "id" => object_id,
 412       "published" => datetime,
 413       "sensitive" => false,
 414       "summary" => "",
 415       "tag" => [],
 416       "to" => ["https://www.w3.org/ns/activitystreams#Public"],
 417       "type" => "Note"
 418     }
 419
 420     {activity_data, object_data}
 421   end
 422
 423   defp prepare_activity_data(actor, "public", _mention) do
 424     obj_data = %{
 425       "cc" => [actor.follower_address],
 426       "to" => [Constants.as_public()]
 427     }
 428
 429     act_data = %{
 430       "cc" => [actor.follower_address],
 431       "to" => [Constants.as_public()]
 432     }
 433
 434     {act_data, obj_data}
 435   end
 436
 437   defp prepare_activity_data(actor, "private", _mention) do
 438     obj_data = %{
 439       "cc" => [],
 440       "to" => [actor.follower_address]
 441     }
 442
 443     act_data = %{
 444       "cc" => [],
 445       "to" => [actor.follower_address]
 446     }
 447
 448     {act_data, obj_data}
 449   end
 450
 451   defp prepare_activity_data(actor, "unlisted", _mention) do
 452     obj_data = %{
 453       "cc" => [Constants.as_public()],
 454       "to" => [actor.follower_address]
 455     }
 456
 457     act_data = %{
 458       "cc" => [Constants.as_public()],
 459       "to" => [actor.follower_address]
 460     }
 461
 462     {act_data, obj_data}
 463   end
 464
 465   defp prepare_activity_data(_actor, "direct", mention) do
 466     %{host: mentioned_host} = URI.parse(mention.ap_id)
 467
 468     obj_data = %{
 469       "cc" => [],
 470       "content" =>
 471         "<span class=\"h-card\"><a class=\"u-url mention\" href=\"#{mention.ap_id}\" rel=\"ugc\">@<span>#{
 472           mention.nickname
 473         }</span></a></span> direct message",
 474       "tag" => [
 475         %{
 476           "href" => mention.ap_id,
 477           "name" => "@#{mention.nickname}@#{mentioned_host}",
 478           "type" => "Mention"
 479         }
 480       ],
 481       "to" => [mention.ap_id]
 482     }
 483
 484     act_data = %{
 485       "cc" => [],
 486       "directMessage" => true,
 487       "to" => [mention.ap_id]
 488     }
 489
 490     {act_data, obj_data}
 491   end
 492
 493   defp get_reply_tasks("public", :user) do
 494     [:friends_local, :friends_remote, :non_friends_local, :non_friends_remote, :user]
 495   end
 496
 497   defp get_reply_tasks("public", group) when group in @friends_groups do
 498     [:non_friends_local, :non_friends_remote, :user, :friends_local, :friends_remote]
 499   end
 500
 501   defp get_reply_tasks("public", group) when group in @non_friends_groups do
 502     [:user, :friends_local, :friends_remote, :non_friends_local, :non_friends_remote]
 503   end
 504
 505   defp get_reply_tasks(visibility, :user) when visibility in ["unlisted", "private"] do
 506     [:friends_local, :friends_remote, :user, :friends_local, :friends_remote]
 507   end
 508
 509   defp get_reply_tasks(visibility, group)
 510        when visibility in ["unlisted", "private"] and group in @friends_groups do
 511     [:user, :friends_remote, :friends_local, :user]
 512   end
 513
 514   defp get_reply_tasks(visibility, group)
 515        when visibility in ["unlisted", "private"] and
 516               group in @non_friends_groups,
 517        do: []
 518
 519   defp get_reply_tasks("direct", :user), do: [:friends_local, :user, :friends_remote]
 520
 521   defp get_reply_tasks("direct", group) when group in @friends_groups,
 522     do: [:user, group, :user]
 523
 524   defp get_reply_tasks("direct", group) when group in @non_friends_groups do
 525     [:user, :non_friends_remote, :user, :non_friends_local]
 526   end
 527
 528   defp insert_replies(tasks, visibility, users, acc) do
 529     Enum.reduce(tasks, acc, fn
 530       :user, {id, data} ->
 531         insert_reply(users[:user], data, id, visibility)
 532
 533       group, {id, data} ->
 534         replier = Enum.random(users[group])
 535         insert_reply(replier, data, id, visibility)
 536     end)
 537   end
 538
 539   defp insert_direct_replies(tasks, user, list, acc) do
 540     Enum.reduce(tasks, acc, fn
 541       :user, {id, data} ->
 542         {reply_id, _} = insert_reply(user, List.delete(data, "@" <> user.nickname), id, "direct")
 543         {reply_id, data}
 544
 545       _, {id, data} ->
 546         actor = Enum.random(list)
 547
 548         {reply_id, _} =
 549           insert_reply(actor, List.delete(data, "@" <> actor.nickname), id, "direct")
 550
 551         {reply_id, data}
 552     end)
 553   end
 554
 555   defp insert_reply(actor, data, activity_id, visibility) do
 556     {:ok, reply} =
 557       CommonAPI.post(actor, %{
 558         status: Enum.join(data, ", "),
 559         visibility: visibility,
 560         in_reply_to_status_id: activity_id
 561       })
 562
 563     {reply.id, ["@" <> actor.nickname | data]}
 564   end
 565
 566   defp get_random_mentions(_users, count) when count == 0, do: []
 567
 568   defp get_random_mentions(users, count) do
 569     users
 570     |> Enum.shuffle()
 571     |> Enum.take(count)
 572     |> Enum.map(&"@#{&1.nickname}")
 573   end
 574
 575   defp get_random_create_activity_id do
 576     Repo.one(
 577       from(a in Pleroma.Activity,
 578         where: fragment("(?)->>'type' = ?", a.data, ^"Create"),
 579         order_by: fragment("RANDOM()"),
 580         limit: 1,
 581         select: a.id
 582       )
 583     )
 584   end
 585 end