(args: RegisterArgs, vm: &VirtualMachine)
| 1117 | #[cfg(unix)] |
| 1118 | #[pyfunction] |
| 1119 | fn register(args: RegisterArgs, vm: &VirtualMachine) -> PyResult<()> { |
| 1120 | check_signum(args.signum, vm)?; |
| 1121 | |
| 1122 | let fd = get_fd_from_file_opt(args.file, vm)?; |
| 1123 | |
| 1124 | let signum = args.signum as usize; |
| 1125 | |
| 1126 | // Get current handler to save as previous |
| 1127 | let previous = if !user_signals::is_enabled(signum) { |
| 1128 | unsafe { |
| 1129 | let mut action: libc::sigaction = core::mem::zeroed(); |
| 1130 | action.sa_sigaction = faulthandler_user_signal as *const () as libc::sighandler_t; |
| 1131 | // SA_RESTART by default; SA_NODEFER only when chaining |
| 1132 | // (faulthandler.c:860-864) |
| 1133 | action.sa_flags = if args.chain { |
| 1134 | libc::SA_NODEFER |
| 1135 | } else { |
| 1136 | libc::SA_RESTART |
| 1137 | }; |
| 1138 | |
| 1139 | let mut prev: libc::sigaction = core::mem::zeroed(); |
| 1140 | if libc::sigaction(args.signum, &action, &mut prev) != 0 { |
| 1141 | return Err(vm.new_os_error(format!( |
| 1142 | "Failed to register signal handler for signal {}", |
| 1143 | args.signum |
| 1144 | ))); |
| 1145 | } |
| 1146 | prev |
| 1147 | } |
| 1148 | } else { |
| 1149 | // Already registered, keep previous handler |
| 1150 | user_signals::get_user_signal(signum) |
| 1151 | .map(|u| u.previous) |
| 1152 | .unwrap_or(unsafe { core::mem::zeroed() }) |
| 1153 | }; |
| 1154 | |
| 1155 | user_signals::set_user_signal( |
| 1156 | signum, |
| 1157 | user_signals::UserSignal { |
| 1158 | enabled: true, |
| 1159 | fd, |
| 1160 | all_threads: args.all_threads, |
| 1161 | chain: args.chain, |
| 1162 | previous, |
| 1163 | }, |
| 1164 | ); |
| 1165 | |
| 1166 | Ok(()) |
| 1167 | } |
| 1168 | |
| 1169 | #[cfg(unix)] |
| 1170 | #[pyfunction] |
nothing calls this directly
no test coverage detected