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