Skip to content

Commit baf60ce

Browse files
changlei-lichanglei-li
committed
CA-422282 Fix race condition about xenops cache
There is race condition about vm cache between pool_migrate_complete and VM event. In the cross-pool migration case, it is designed to create vm with power_state Halted in XAPI db. In pool_migrate_complete, add_caches create an empty xenops_chae for the VM, then refresh_vm compares the cache powerstate None with its real state Running to update the right powerstate to XAPI db. In the fail case, it is found that: -> VM event 1 update_vm -> pool_migrate_complete add_caches (cache power_state None) -> pool_migrate_complete refresh_vm -> VM event 1 update cache (cache power_state Running) -> VM event 2 update_vm (Running <-> Running, XAPI DB not update) When pool_migrate_complete add_caches, the cache update of previous VM event 1 breaks the design intention. This PR Add a new interface add_caches_and_refresh to make the procedure race-free. 1. Wait for barrier - ensure all in-flight events complete 2. Suppress + add_caches - prevent new events 3. Refresh Signed-off-by: Changlei Li <changlei.li@cloud.com>
1 parent 1878ccc commit baf60ce

2 files changed

Lines changed: 118 additions & 107 deletions

File tree

‎ocaml/xapi/xapi_vm_migrate.ml‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -492,8 +492,7 @@ let pool_migrate_complete ~__context ~vm ~host:_ =
492492
if Xapi_xenops.vm_exists_in_xenopsd queue_name dbg id then (
493493
remove_stale_pcis ~__context ~vm ;
494494
Xapi_xenops.set_resident_on ~__context ~self:vm ;
495-
Xapi_xenops.add_caches id ;
496-
Xapi_xenops.refresh_vm ~__context ~self:vm ;
495+
Xapi_xenops.add_caches_and_refresh queue_name dbg id ;
497496
Monitor_dbcalls_cache.clear_cache_for_vm ~vm_uuid:id
498497
) ;
499498
(* Reset the state, which will update allowed operations, clear reservations

‎ocaml/xapi/xapi_xenops.ml‎

Lines changed: 117 additions & 105 deletions
Original file line numberDiff line numberDiff line change
@@ -1494,6 +1494,111 @@ let string_of_exn = function
14941494
| e ->
14951495
Printexc.to_string e
14961496

1497+
module Events_from_xenopsd = struct
1498+
type t = {mutable finished: bool; m: Mutex.t; c: Condition.t}
1499+
1500+
let make () = {finished= false; m= Mutex.create (); c= Condition.create ()}
1501+
1502+
let active = Hashtbl.create 10
1503+
1504+
let active_m = Mutex.create ()
1505+
1506+
let register =
1507+
let counter = ref 0 in
1508+
fun t ->
1509+
with_lock active_m (fun () ->
1510+
let id = !counter in
1511+
incr counter ;
1512+
Hashtbl.replace active id t ;
1513+
id
1514+
)
1515+
1516+
let wait queue_name dbg vm_id () =
1517+
let module Client = (val make_client queue_name : XENOPS) in
1518+
let t = make () in
1519+
let id = register t in
1520+
Debug_info.with_dbg
1521+
~attributes:
1522+
[
1523+
("messaging.operation.name", "subscribe")
1524+
; ("messaging.system", "event")
1525+
; ("messaging.destination.subscription.name", vm_id)
1526+
; ("messaging.message.id", string_of_int id)
1527+
]
1528+
~name:("subscribe" ^ " " ^ queue_name)
1529+
~dbg
1530+
@@ fun di ->
1531+
let dbg = Debug_info.to_string di in
1532+
debug "Client.UPDATES.inject_barrier %d" id ;
1533+
Client.UPDATES.inject_barrier dbg vm_id id ;
1534+
with_lock t.m (fun () ->
1535+
while not t.finished do
1536+
Condition.wait t.c t.m
1537+
done
1538+
)
1539+
1540+
let wakeup queue_name dbg id =
1541+
Debug_info.with_dbg
1542+
~attributes:
1543+
[
1544+
("messaging.operation.name", "settle")
1545+
; ("messaging.system", "event")
1546+
; ("messaging.message.id", string_of_int id)
1547+
]
1548+
~name:("settle" ^ " " ^ queue_name)
1549+
~dbg
1550+
@@ fun di ->
1551+
let dbg = Debug_info.to_string di in
1552+
let module Client = (val make_client queue_name : XENOPS) in
1553+
Client.UPDATES.remove_barrier dbg id ;
1554+
let t =
1555+
with_lock active_m @@ fun () ->
1556+
match Hashtbl.find_opt active id with
1557+
| Some t ->
1558+
Hashtbl.remove active id ; Some t
1559+
| None ->
1560+
warn "Events_from_xenopsd.wakeup: unknown id %d" id ;
1561+
None
1562+
in
1563+
Option.iter
1564+
(fun t ->
1565+
with_lock t.m @@ fun () ->
1566+
t.finished <- true ;
1567+
Condition.signal t.c
1568+
)
1569+
t
1570+
1571+
let events_suppressed_on = Hashtbl.create 10
1572+
1573+
let events_suppressed_on_m = Mutex.create ()
1574+
1575+
let events_suppressed_on_c = Condition.create ()
1576+
1577+
let are_suppressed vm = Hashtbl.mem events_suppressed_on vm
1578+
1579+
let with_suppressed queue_name dbg vm_id f =
1580+
debug "suppressing xenops events on VM: %s" vm_id ;
1581+
let module Client = (val make_client queue_name : XENOPS) in
1582+
with_lock events_suppressed_on_m (fun () ->
1583+
Hashtbl.add events_suppressed_on vm_id ()
1584+
) ;
1585+
finally f (fun () ->
1586+
with_lock events_suppressed_on_m (fun () ->
1587+
Hashtbl.remove events_suppressed_on vm_id ;
1588+
if not (Hashtbl.mem events_suppressed_on vm_id) then (
1589+
debug "re-enabled xenops events on VM: %s; refreshing VM" vm_id ;
1590+
Client.UPDATES.refresh_vm dbg vm_id ;
1591+
wait queue_name dbg vm_id () ;
1592+
Condition.broadcast events_suppressed_on_c
1593+
) else
1594+
while are_suppressed vm_id do
1595+
debug "waiting for events to become re-enabled" ;
1596+
Condition.wait events_suppressed_on_c events_suppressed_on_m
1597+
done
1598+
)
1599+
)
1600+
end
1601+
14971602
(* Serialise updates to the metadata caches *)
14981603
let metadata_m = Mutex.create ()
14991604

@@ -1819,6 +1924,18 @@ let add_caches id =
18191924
Xenops_cache.register id
18201925
)
18211926

1927+
let add_caches_and_refresh queue_name dbg id =
1928+
(* Fix race condition with proper three-step approach:
1929+
Step 1: Wait for barrier - ensures all in-flight events complete
1930+
This prevents TOCTOU where an event passes the suppression check
1931+
but writes to cache after the wipe.
1932+
Step 2: Suppress + wipe - prevents new events during cache wipe
1933+
Step 3: Refresh (automatic) - with_suppressed cleanup calls refresh_vm
1934+
which generates fresh events and waits for barrier
1935+
*)
1936+
Events_from_xenopsd.wait queue_name dbg id () ;
1937+
Events_from_xenopsd.with_suppressed queue_name dbg id (fun () -> add_caches id)
1938+
18221939
let to_xenops_console_protocol =
18231940
let open Vm in
18241941
function `rfb -> Rfb | `vt100 -> Vt100 | `rdp -> Rfb
@@ -1842,111 +1959,6 @@ let to_xenapi_console_protocol =
18421959
let trigger_xenapi_reregister =
18431960
ref (fun () -> debug "No xapi event thread to wake up")
18441961

1845-
module Events_from_xenopsd = struct
1846-
type t = {mutable finished: bool; m: Mutex.t; c: Condition.t}
1847-
1848-
let make () = {finished= false; m= Mutex.create (); c= Condition.create ()}
1849-
1850-
let active = Hashtbl.create 10
1851-
1852-
let active_m = Mutex.create ()
1853-
1854-
let register =
1855-
let counter = ref 0 in
1856-
fun t ->
1857-
with_lock active_m (fun () ->
1858-
let id = !counter in
1859-
incr counter ;
1860-
Hashtbl.replace active id t ;
1861-
id
1862-
)
1863-
1864-
let wait queue_name dbg vm_id () =
1865-
let module Client = (val make_client queue_name : XENOPS) in
1866-
let t = make () in
1867-
let id = register t in
1868-
Debug_info.with_dbg
1869-
~attributes:
1870-
[
1871-
("messaging.operation.name", "subscribe")
1872-
; ("messaging.system", "event")
1873-
; ("messaging.destination.subscription.name", vm_id)
1874-
; ("messaging.message.id", string_of_int id)
1875-
]
1876-
~name:("subscribe" ^ " " ^ queue_name)
1877-
~dbg
1878-
@@ fun di ->
1879-
let dbg = Debug_info.to_string di in
1880-
debug "Client.UPDATES.inject_barrier %d" id ;
1881-
Client.UPDATES.inject_barrier dbg vm_id id ;
1882-
with_lock t.m (fun () ->
1883-
while not t.finished do
1884-
Condition.wait t.c t.m
1885-
done
1886-
)
1887-
1888-
let wakeup queue_name dbg id =
1889-
Debug_info.with_dbg
1890-
~attributes:
1891-
[
1892-
("messaging.operation.name", "settle")
1893-
; ("messaging.system", "event")
1894-
; ("messaging.message.id", string_of_int id)
1895-
]
1896-
~name:("settle" ^ " " ^ queue_name)
1897-
~dbg
1898-
@@ fun di ->
1899-
let dbg = Debug_info.to_string di in
1900-
let module Client = (val make_client queue_name : XENOPS) in
1901-
Client.UPDATES.remove_barrier dbg id ;
1902-
let t =
1903-
with_lock active_m @@ fun () ->
1904-
match Hashtbl.find_opt active id with
1905-
| Some t ->
1906-
Hashtbl.remove active id ; Some t
1907-
| None ->
1908-
warn "Events_from_xenopsd.wakeup: unknown id %d" id ;
1909-
None
1910-
in
1911-
Option.iter
1912-
(fun t ->
1913-
with_lock t.m @@ fun () ->
1914-
t.finished <- true ;
1915-
Condition.signal t.c
1916-
)
1917-
t
1918-
1919-
let events_suppressed_on = Hashtbl.create 10
1920-
1921-
let events_suppressed_on_m = Mutex.create ()
1922-
1923-
let events_suppressed_on_c = Condition.create ()
1924-
1925-
let are_suppressed vm = Hashtbl.mem events_suppressed_on vm
1926-
1927-
let with_suppressed queue_name dbg vm_id f =
1928-
debug "suppressing xenops events on VM: %s" vm_id ;
1929-
let module Client = (val make_client queue_name : XENOPS) in
1930-
with_lock events_suppressed_on_m (fun () ->
1931-
Hashtbl.add events_suppressed_on vm_id ()
1932-
) ;
1933-
finally f (fun () ->
1934-
with_lock events_suppressed_on_m (fun () ->
1935-
Hashtbl.remove events_suppressed_on vm_id ;
1936-
if not (Hashtbl.mem events_suppressed_on vm_id) then (
1937-
debug "re-enabled xenops events on VM: %s; refreshing VM" vm_id ;
1938-
Client.UPDATES.refresh_vm dbg vm_id ;
1939-
wait queue_name dbg vm_id () ;
1940-
Condition.broadcast events_suppressed_on_c
1941-
) else
1942-
while are_suppressed vm_id do
1943-
debug "waiting for events to become re-enabled" ;
1944-
Condition.wait events_suppressed_on_c events_suppressed_on_m
1945-
done
1946-
)
1947-
)
1948-
end
1949-
19501962
let update_vm_internal ~__context ~id ~self ~previous ~info ~localhost =
19511963
debug "xenopsd event: processing event for VM %s" id ;
19521964
if info = None then

0 commit comments

Comments
 (0)