Skip to content

Commit 012c741

Browse files
anaPerezGhigliaanaPerezGhiglia
authored
#1744: fix respondent_stats deadlocks (#2327)
* create task for triggering deadlocks (pun intended :P ) * Create migration for fixing respondents_upd trigger * Update structure.sql after migration
1 parent 89a24d0 commit 012c741

File tree

4 files changed

+218
-34
lines changed

4 files changed

+218
-34
lines changed

‎lib/mix/tasks/deadlock.test.ex

+61
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
defmodule Mix.Tasks.Ask.Deadlock.Test do
2+
import Ecto.Query
3+
use Mix.Task
4+
5+
alias Ask.{Repo, Respondent}
6+
7+
8+
@shortdoc """
9+
Edits multiple respondents at the same time to try to generate a deadlock by respondent_stats locks.
10+
"""
11+
12+
@impl Mix.Task
13+
def run([survey_id]) do
14+
run([survey_id, :contacted, :queued])
15+
end
16+
17+
# Prerequisites
18+
# have a survey with respondents
19+
# - state:active, disposition:disposition_a
20+
# - state:active, disposition:disposition_b
21+
#
22+
# Steps to achieve that:
23+
# - create new survey and load respondent sample (without launching)
24+
# - update n respondents to be active & disposition_a
25+
# - update m respondents to be active & disposition_b
26+
def run([survey_id, disposition_a, disposition_b]) do
27+
Mix.shell().info("Starting deadlock test..")
28+
Mix.shell().info("Max concurrency: #{System.schedulers_online()}")
29+
Mix.shell().info("Updating respondents from dispositions #{disposition_a} <-> #{disposition_b}")
30+
31+
Mix.Task.run("app.start")
32+
33+
disposition_a_respondents = Repo.all(from r in Respondent,
34+
where: r.survey_id == ^survey_id,
35+
where: r.state == :active,
36+
where: r.disposition == ^disposition_a
37+
)
38+
39+
disposition_b_respondents = Repo.all(from r in Respondent,
40+
where: r.survey_id == ^survey_id,
41+
where: r.state == :active,
42+
where: r.disposition == ^disposition_b
43+
)
44+
45+
tasks1 = respondents_to_update_tasks(disposition_a_respondents, disposition_b)
46+
47+
tasks2 = respondents_to_update_tasks(disposition_b_respondents, disposition_a)
48+
49+
Task.yield_many(tasks1 ++ tasks2, :infinity)
50+
51+
end
52+
53+
def respondents_to_update_tasks(respondents, new_disposition) do
54+
respondents
55+
|> Enum.map(fn r ->
56+
Task.async(fn ->
57+
Respondent.update(r, %{disposition: new_disposition}, true)
58+
end)
59+
end)
60+
end
61+
end

‎mix.exs

+2-1
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,8 @@ defmodule Ask.Mixfile do
116116
"ecto.reset": ["ecto.drop", "ecto.setup"],
117117
"ecto.migrate": ["ecto.migrate", "ask.ecto_dump"],
118118
"ecto.rollback": ["ecto.rollback", "ask.ecto_dump"],
119-
test: ["ecto.create --quiet", "ecto.migrate --quiet", "test"]
119+
test: ["ecto.create --quiet", "ecto.migrate --quiet", "test"],
120+
deadlocks: ["ask.deadlock.test"]
120121
]
121122
end
122123
end

‎priv/repo/migrations/20240422175453_fix_respondent_stats_trigger.exs

+97
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
defmodule Ask.Repo.Migrations.FixRespondentStatsTrigger do
2+
use Ecto.Migration
3+
alias Ask.Repo
4+
5+
def up do
6+
Repo.query!(
7+
"""
8+
DROP TRIGGER respondents_upd;
9+
"""
10+
)
11+
12+
Repo.query!(
13+
"""
14+
CREATE TRIGGER respondents_upd
15+
AFTER UPDATE ON respondents
16+
FOR EACH ROW
17+
BEGIN
18+
19+
# This is a hack for using the `select for update`
20+
# so we can lock all the records that will get updated
21+
# by this trigger before actually modifying them
22+
# so we prevent this trigger to generate deadlocks
23+
# see #1744
24+
DECLARE temp_stats INT;
25+
26+
SELECT count(*)
27+
INTO temp_stats
28+
FROM respondent_stats
29+
WHERE (survey_id = OLD.survey_id
30+
AND questionnaire_id = IFNULL(OLD.questionnaire_id, 0)
31+
AND state = OLD.state
32+
AND disposition = OLD.disposition
33+
AND quota_bucket_id = IFNULL(OLD.quota_bucket_id, 0)
34+
AND mode = IFNULL(OLD.mode, ''))
35+
OR (survey_id = NEW.survey_id
36+
AND questionnaire_id = IFNULL(NEW.questionnaire_id, 0)
37+
AND state = NEW.state
38+
AND disposition = NEW.disposition
39+
AND quota_bucket_id = IFNULL(NEW.quota_bucket_id, 0)
40+
AND mode = IFNULL(NEW.mode, ''))
41+
FOR UPDATE;
42+
43+
UPDATE respondent_stats
44+
SET `count` = `count` - 1
45+
WHERE survey_id = OLD.survey_id
46+
AND questionnaire_id = IFNULL(OLD.questionnaire_id, 0)
47+
AND state = OLD.state
48+
AND disposition = OLD.disposition
49+
AND quota_bucket_id = IFNULL(OLD.quota_bucket_id, 0)
50+
AND mode = IFNULL(OLD.mode, '')
51+
;
52+
53+
INSERT INTO respondent_stats(survey_id, questionnaire_id, state, disposition, quota_bucket_id, mode, `count`)
54+
VALUES (NEW.survey_id, IFNULL(NEW.questionnaire_id, 0), NEW.state, NEW.disposition, IFNULL(NEW.quota_bucket_id, 0), IFNULL(NEW.mode, ''), 1)
55+
ON DUPLICATE KEY UPDATE `count` = `count` + 1
56+
;
57+
58+
END;
59+
"""
60+
)
61+
62+
end
63+
64+
def down do
65+
# Re-create the trigger as it was before
66+
Repo.query!(
67+
"""
68+
DROP TRIGGER respondents_upd;
69+
"""
70+
)
71+
72+
Repo.query!(
73+
"""
74+
CREATE TRIGGER respondents_upd
75+
AFTER UPDATE ON respondents
76+
FOR EACH ROW
77+
BEGIN
78+
79+
UPDATE respondent_stats
80+
SET `count` = `count` - 1
81+
WHERE survey_id = OLD.survey_id
82+
AND questionnaire_id = IFNULL(OLD.questionnaire_id, 0)
83+
AND state = OLD.state
84+
AND disposition = OLD.disposition
85+
AND quota_bucket_id = IFNULL(OLD.quota_bucket_id, 0)
86+
AND mode = IFNULL(OLD.mode, '')
87+
;
88+
89+
INSERT INTO respondent_stats(survey_id, questionnaire_id, state, disposition, quota_bucket_id, mode, `count`)
90+
VALUES (NEW.survey_id, IFNULL(NEW.questionnaire_id, 0), NEW.state, NEW.disposition, IFNULL(NEW.quota_bucket_id, 0), IFNULL(NEW.mode, ''), 1)
91+
ON DUPLICATE KEY UPDATE `count` = `count` + 1
92+
;
93+
94+
END;
95+
""")
96+
end
97+
end

‎priv/repo/structure.sql

+58-33
Original file line numberDiff line numberDiff line change
@@ -602,38 +602,6 @@ DELIMITER ;;
602602
ON DUPLICATE KEY UPDATE `count` = `count` + 1;
603603
END IF;
604604
605-
END */;;
606-
DELIMITER ;
607-
/*!50003 SET sql_mode = @saved_sql_mode */ ;
608-
/*!50003 SET character_set_client = @saved_cs_client */ ;
609-
/*!50003 SET character_set_results = @saved_cs_results */ ;
610-
/*!50003 SET collation_connection = @saved_col_connection */ ;
611-
/*!50003 SET @saved_cs_client = @@character_set_client */ ;
612-
/*!50003 SET @saved_cs_results = @@character_set_results */ ;
613-
/*!50003 SET @saved_col_connection = @@collation_connection */ ;
614-
/*!50003 SET character_set_client = utf8mb3 */ ;
615-
/*!50003 SET character_set_results = utf8mb3 */ ;
616-
/*!50003 SET collation_connection = utf8mb3_general_ci */ ;
617-
/*!50003 SET @saved_sql_mode = @@sql_mode */ ;
618-
/*!50003 SET sql_mode = 'NO_AUTO_VALUE_ON_ZERO' */ ;
619-
DELIMITER ;;
620-
/*!50003 CREATE*/ /*!50017 DEFINER=`root`@`%`*/ /*!50003 TRIGGER `respondents_upd` AFTER UPDATE ON `respondents` FOR EACH ROW BEGIN
621-
622-
UPDATE respondent_stats
623-
SET `count` = `count` - 1
624-
WHERE survey_id = OLD.survey_id
625-
AND questionnaire_id = IFNULL(OLD.questionnaire_id, 0)
626-
AND state = OLD.state
627-
AND disposition = OLD.disposition
628-
AND quota_bucket_id = IFNULL(OLD.quota_bucket_id, 0)
629-
AND mode = IFNULL(OLD.mode, '')
630-
;
631-
632-
INSERT INTO respondent_stats(survey_id, questionnaire_id, state, disposition, quota_bucket_id, mode, `count`)
633-
VALUES (NEW.survey_id, IFNULL(NEW.questionnaire_id, 0), NEW.state, NEW.disposition, IFNULL(NEW.quota_bucket_id, 0), IFNULL(NEW.mode, ''), 1)
634-
ON DUPLICATE KEY UPDATE `count` = `count` + 1
635-
;
636-
637605
END */;;
638606
DELIMITER ;
639607
/*!50003 SET sql_mode = @saved_sql_mode */ ;
@@ -665,6 +633,62 @@ DELIMITER ;;
665633
AND date = DATE(OLD.updated_at);
666634
END IF;
667635
636+
END */;;
637+
DELIMITER ;
638+
/*!50003 SET sql_mode = @saved_sql_mode */ ;
639+
/*!50003 SET character_set_client = @saved_cs_client */ ;
640+
/*!50003 SET character_set_results = @saved_cs_results */ ;
641+
/*!50003 SET collation_connection = @saved_col_connection */ ;
642+
/*!50003 SET @saved_cs_client = @@character_set_client */ ;
643+
/*!50003 SET @saved_cs_results = @@character_set_results */ ;
644+
/*!50003 SET @saved_col_connection = @@collation_connection */ ;
645+
/*!50003 SET character_set_client = utf8mb4 */ ;
646+
/*!50003 SET character_set_results = utf8mb4 */ ;
647+
/*!50003 SET collation_connection = utf8mb4_general_ci */ ;
648+
/*!50003 SET @saved_sql_mode = @@sql_mode */ ;
649+
/*!50003 SET sql_mode = 'ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION' */ ;
650+
DELIMITER ;;
651+
/*!50003 CREATE*/ /*!50017 DEFINER=`root`@`%`*/ /*!50003 TRIGGER `respondents_upd` AFTER UPDATE ON `respondents` FOR EACH ROW BEGIN
652+
653+
# This is a hack for using the `select for update`
654+
# so we can lock all the records that will get updated
655+
# by this trigger before actually modifying them
656+
# so we prevent this trigger to generate deadlocks
657+
# see #1744
658+
DECLARE temp_stats INT;
659+
660+
SELECT count(*)
661+
INTO temp_stats
662+
FROM respondent_stats
663+
WHERE (survey_id = OLD.survey_id
664+
AND questionnaire_id = IFNULL(OLD.questionnaire_id, 0)
665+
AND state = OLD.state
666+
AND disposition = OLD.disposition
667+
AND quota_bucket_id = IFNULL(OLD.quota_bucket_id, 0)
668+
AND mode = IFNULL(OLD.mode, ''))
669+
OR (survey_id = NEW.survey_id
670+
AND questionnaire_id = IFNULL(NEW.questionnaire_id, 0)
671+
AND state = NEW.state
672+
AND disposition = NEW.disposition
673+
AND quota_bucket_id = IFNULL(NEW.quota_bucket_id, 0)
674+
AND mode = IFNULL(NEW.mode, ''))
675+
FOR UPDATE;
676+
677+
UPDATE respondent_stats
678+
SET `count` = `count` - 1
679+
WHERE survey_id = OLD.survey_id
680+
AND questionnaire_id = IFNULL(OLD.questionnaire_id, 0)
681+
AND state = OLD.state
682+
AND disposition = OLD.disposition
683+
AND quota_bucket_id = IFNULL(OLD.quota_bucket_id, 0)
684+
AND mode = IFNULL(OLD.mode, '')
685+
;
686+
687+
INSERT INTO respondent_stats(survey_id, questionnaire_id, state, disposition, quota_bucket_id, mode, `count`)
688+
VALUES (NEW.survey_id, IFNULL(NEW.questionnaire_id, 0), NEW.state, NEW.disposition, IFNULL(NEW.quota_bucket_id, 0), IFNULL(NEW.mode, ''), 1)
689+
ON DUPLICATE KEY UPDATE `count` = `count` + 1
690+
;
691+
668692
END */;;
669693
DELIMITER ;
670694
/*!50003 SET sql_mode = @saved_sql_mode */ ;
@@ -994,7 +1018,7 @@ CREATE TABLE `users` (
9941018
/*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */;
9951019
/*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
9961020

997-
-- Dump completed on 2024-02-21 5:26:53
1021+
-- Dump completed on 2024-04-22 22:06:45
9981022
INSERT INTO `schema_migrations` (version) VALUES (20160812145257);
9991023
INSERT INTO `schema_migrations` (version) VALUES (20160816183915);
10001024
INSERT INTO `schema_migrations` (version) VALUES (20160830200454);
@@ -1213,3 +1237,4 @@ INSERT INTO `schema_migrations` (version) VALUES (20230402091100);
12131237
INSERT INTO `schema_migrations` (version) VALUES (20230405111657);
12141238
INSERT INTO `schema_migrations` (version) VALUES (20230413101342);
12151239
INSERT INTO `schema_migrations` (version) VALUES (20230821100203);
1240+
INSERT INTO `schema_migrations` (version) VALUES (20240422175453);

0 commit comments

Comments
 (0)