1 use alloc::{ 2 collections::LinkedList, 3 sync::{Arc, Weak}, 4 }; 5 use core::hash::{Hash, Hasher}; 6 use core::{intrinsics::likely, sync::atomic::AtomicU64}; 7 use hashbrown::HashMap; 8 use system_error::SystemError; 9 10 use crate::{ 11 arch::{sched::sched, CurrentIrqArch, MMArch}, 12 exception::InterruptArch, 13 libs::spinlock::{SpinLock, SpinLockGuard}, 14 mm::{ucontext::AddressSpace, MemoryManagementArch, VirtAddr}, 15 process::{ProcessControlBlock, ProcessManager}, 16 syscall::user_access::UserBufferReader, 17 time::{ 18 timer::{next_n_us_timer_jiffies, Timer, WakeUpHelper}, 19 TimeSpec, 20 }, 21 }; 22 23 use super::constant::*; 24 25 static mut FUTEX_DATA: Option<FutexData> = None; 26 27 pub struct FutexData { 28 data: SpinLock<HashMap<FutexKey, FutexHashBucket>>, 29 } 30 31 impl FutexData { futex_map() -> SpinLockGuard<'static, HashMap<FutexKey, FutexHashBucket>>32 pub fn futex_map() -> SpinLockGuard<'static, HashMap<FutexKey, FutexHashBucket>> { 33 unsafe { FUTEX_DATA.as_ref().unwrap().data.lock() } 34 } 35 try_remove(key: &FutexKey) -> Option<FutexHashBucket>36 pub fn try_remove(key: &FutexKey) -> Option<FutexHashBucket> { 37 unsafe { 38 let mut guard = FUTEX_DATA.as_ref().unwrap().data.lock(); 39 if let Some(futex) = guard.get(key) { 40 if futex.chain.is_empty() { 41 return guard.remove(key); 42 } 43 } 44 } 45 None 46 } 47 } 48 49 pub struct Futex; 50 51 // 对于同一个futex的进程或线程将会在这个bucket等待 52 pub struct FutexHashBucket { 53 // 该futex维护的等待队列 54 chain: LinkedList<Arc<FutexObj>>, 55 } 56 57 impl FutexHashBucket { 58 /// ## 判断是否在bucket里 contains(&self, futex_q: &FutexObj) -> bool59 pub fn contains(&self, futex_q: &FutexObj) -> bool { 60 self.chain 61 .iter() 62 .filter(|x| futex_q.pcb.ptr_eq(&x.pcb) && x.key == futex_q.key) 63 .count() 64 != 0 65 } 66 67 /// 让futex_q在该bucket上挂起 68 /// 69 /// 进入该函数前，需要关中断 70 #[inline(always)] sleep_no_sched(&mut self, futex_q: Arc<FutexObj>) -> Result<(), SystemError>71 pub fn sleep_no_sched(&mut self, futex_q: Arc<FutexObj>) -> Result<(), SystemError> { 72 assert!(CurrentIrqArch::is_irq_enabled() == false); 73 self.chain.push_back(futex_q); 74 75 ProcessManager::mark_sleep(true)?; 76 77 Ok(()) 78 } 79 80 /// ## 唤醒队列中的最多nr_wake个进程 81 /// 82 /// return: 唤醒的进程数 83 #[inline(always)] wake_up( &mut self, key: FutexKey, bitset: Option<u32>, nr_wake: u32, ) -> Result<usize, SystemError>84 pub fn wake_up( 85 &mut self, 86 key: FutexKey, 87 bitset: Option<u32>, 88 nr_wake: u32, 89 ) -> Result<usize, SystemError> { 90 let mut count = 0; 91 let mut pop_count = 0; 92 while let Some(futex_q) = self.chain.pop_front() { 93 if futex_q.key == key { 94 // TODO: 考虑优先级继承的机制 95 96 if let Some(bitset) = bitset { 97 if futex_q.bitset != bitset { 98 self.chain.push_back(futex_q); 99 continue; 100 } 101 } 102 103 // 唤醒 104 if futex_q.pcb.upgrade().is_some() { 105 self.remove(futex_q.clone()); 106 ProcessManager::wakeup(&futex_q.pcb.upgrade().unwrap())?; 107 } 108 109 // 判断唤醒数 110 count += 1; 111 if count >= nr_wake { 112 break; 113 } 114 } else { 115 self.chain.push_back(futex_q); 116 } 117 // 判断是否循环完队列了 118 pop_count += 1; 119 if pop_count >= self.chain.len() { 120 break; 121 } 122 } 123 Ok(count as usize) 124 } 125 126 /// 将FutexObj从bucket中删除 remove(&mut self, futex: Arc<FutexObj>)127 pub fn remove(&mut self, futex: Arc<FutexObj>) { 128 self.chain 129 .extract_if(|x| Arc::ptr_eq(x, &futex)) 130 .for_each(drop); 131 } 132 } 133 134 #[derive(Debug)] 135 pub struct FutexObj { 136 pcb: Weak<ProcessControlBlock>, 137 key: FutexKey, 138 bitset: u32, 139 // TODO: 优先级继承 140 } 141 142 pub enum FutexAccess { 143 FutexRead, 144 FutexWrite, 145 } 146 147 #[allow(dead_code)] 148 #[derive(Hash, PartialEq, Eq, Clone, Debug)] 149 /// ### 用于定位内核唯一的futex 150 pub enum InnerFutexKey { 151 Shared(SharedKey), 152 Private(PrivateKey), 153 } 154 155 #[derive(Hash, PartialEq, Eq, Clone, Debug)] 156 pub struct FutexKey { 157 ptr: u64, 158 word: u64, 159 offset: u32, 160 key: InnerFutexKey, 161 } 162 163 /// 不同进程间通过文件共享futex变量，表明该变量在文件中的位置 164 #[derive(Hash, PartialEq, Eq, Clone, Debug)] 165 pub struct SharedKey { 166 i_seq: u64, 167 page_offset: u64, 168 } 169 170 /// 同一进程的不同线程共享futex变量，表明该变量在进程地址空间中的位置 171 #[derive(Clone, Debug)] 172 pub struct PrivateKey { 173 // 所在的地址空间 174 address_space: Option<Weak<AddressSpace>>, 175 // 表示所在页面的初始地址 176 address: u64, 177 } 178 179 impl Hash for PrivateKey { hash<H: Hasher>(&self, state: &mut H)180 fn hash<H: Hasher>(&self, state: &mut H) { 181 self.address.hash(state); 182 } 183 } 184 185 impl Eq for PrivateKey {} 186 187 impl PartialEq for PrivateKey { eq(&self, other: &Self) -> bool188 fn eq(&self, other: &Self) -> bool { 189 if self.address_space.is_none() && other.address_space.is_none() { 190 return self.address == other.address; 191 } else { 192 return self 193 .address_space 194 .as_ref() 195 .unwrap_or(&Weak::default()) 196 .ptr_eq(&other.address_space.as_ref().unwrap_or(&Weak::default())) 197 && self.address == other.address; 198 } 199 } 200 } 201 202 impl Futex { 203 /// ### 初始化FUTEX_DATA init()204 pub fn init() { 205 unsafe { 206 FUTEX_DATA = Some(FutexData { 207 data: SpinLock::new(HashMap::new()), 208 }) 209 }; 210 } 211 212 /// ### 让当前进程在指定futex上等待直到futex_wake显式唤醒 futex_wait( uaddr: VirtAddr, flags: FutexFlag, val: u32, abs_time: Option<TimeSpec>, bitset: u32, ) -> Result<usize, SystemError>213 pub fn futex_wait( 214 uaddr: VirtAddr, 215 flags: FutexFlag, 216 val: u32, 217 abs_time: Option<TimeSpec>, 218 bitset: u32, 219 ) -> Result<usize, SystemError> { 220 if bitset == 0 { 221 return Err(SystemError::EINVAL); 222 } 223 224 // 获取全局hash表的key值 225 let key = Self::get_futex_key( 226 uaddr, 227 flags.contains(FutexFlag::FLAGS_SHARED), 228 FutexAccess::FutexRead, 229 )?; 230 231 let mut futex_map_guard = FutexData::futex_map(); 232 let bucket = futex_map_guard.get_mut(&key); 233 let bucket_mut = match bucket { 234 Some(bucket) => bucket, 235 None => { 236 let bucket = FutexHashBucket { 237 chain: LinkedList::new(), 238 }; 239 futex_map_guard.insert(key.clone(), bucket); 240 futex_map_guard.get_mut(&key).unwrap() 241 } 242 }; 243 244 // 使用UserBuffer读取futex 245 let user_reader = 246 UserBufferReader::new(uaddr.as_ptr::<u32>(), core::mem::size_of::<u32>(), true)?; 247 248 // 从用户空间读取到futex的val 249 let mut uval = 0; 250 251 // 读取 252 // 这里只尝试一种方式去读取用户空间，与linux不太一致 253 // 对于linux，如果bucket被锁住时读取失败，将会将bucket解锁后重新读取 254 user_reader.copy_one_from_user::<u32>(&mut uval, 0)?; 255 256 // 不满足wait条件，返回错误 257 if uval != val { 258 return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); 259 } 260 261 let pcb = ProcessManager::current_pcb(); 262 // 创建超时计时器任务 263 let mut timer = None; 264 if !abs_time.is_none() { 265 let time = abs_time.unwrap(); 266 267 let wakeup_helper = WakeUpHelper::new(pcb.clone()); 268 269 let sec = time.tv_sec; 270 let nsec = time.tv_nsec; 271 let jiffies = next_n_us_timer_jiffies((nsec / 1000 + sec * 1_000_000) as u64); 272 273 let wake_up = Timer::new(wakeup_helper, jiffies); 274 275 wake_up.activate(); 276 timer = Some(wake_up); 277 } 278 279 let futex_q = Arc::new(FutexObj { 280 pcb: Arc::downgrade(&pcb), 281 key: key.clone(), 282 bitset, 283 }); 284 let irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() }; 285 // 满足条件则将当前进程在该bucket上挂起 286 bucket_mut.sleep_no_sched(futex_q.clone()).map_err(|e| { 287 kwarn!("error:{e:?}"); 288 e 289 })?; 290 drop(futex_map_guard); 291 drop(irq_guard); 292 sched(); 293 294 // 被唤醒后的检查 295 let mut futex_map_guard = FutexData::futex_map(); 296 let bucket = futex_map_guard.get_mut(&key); 297 let bucket_mut = match bucket { 298 // 如果该pcb不在链表里面了或者该链表已经被释放，就证明是正常的Wake操作 299 Some(bucket_mut) => { 300 if !bucket_mut.contains(&futex_q) { 301 // 取消定时器任务 302 if timer.is_some() { 303 timer.unwrap().cancel(); 304 } 305 return Ok(0); 306 } 307 // 非正常唤醒，返回交给下层 308 bucket_mut 309 } 310 None => { 311 // 取消定时器任务 312 if timer.is_some() { 313 timer.unwrap().cancel(); 314 } 315 return Ok(0); 316 } 317 }; 318 319 // 如果是超时唤醒，则返回错误 320 if timer.is_some() { 321 if timer.clone().unwrap().timeout() { 322 bucket_mut.remove(futex_q); 323 324 return Err(SystemError::ETIMEDOUT); 325 } 326 } 327 328 // TODO: 如果没有挂起的信号，则重新判断是否满足wait要求，重新进入wait 329 330 // 经过前面的几个判断，到这里之后， 331 // 当前进程被唤醒大概率是其他进程更改了uval,需要重新去判断当前进程是否满足wait 332 333 // 到这里之后，前面的唤醒条件都不满足，则是被信号唤醒 334 // 需要处理信号然后重启futex系统调用 335 336 // 取消定时器任务 337 if timer.is_some() { 338 let timer = timer.unwrap(); 339 if !timer.timeout() { 340 timer.cancel(); 341 } 342 } 343 Ok(0) 344 } 345 346 // ### 唤醒指定futex上挂起的最多nr_wake个进程 futex_wake( uaddr: VirtAddr, flags: FutexFlag, nr_wake: u32, bitset: u32, ) -> Result<usize, SystemError>347 pub fn futex_wake( 348 uaddr: VirtAddr, 349 flags: FutexFlag, 350 nr_wake: u32, 351 bitset: u32, 352 ) -> Result<usize, SystemError> { 353 if bitset == 0 { 354 return Err(SystemError::EINVAL); 355 } 356 357 // 获取futex_key,并且判断地址空间合法性 358 let key = Self::get_futex_key( 359 uaddr, 360 flags.contains(FutexFlag::FLAGS_SHARED), 361 FutexAccess::FutexRead, 362 )?; 363 let mut binding = FutexData::futex_map(); 364 let bucket_mut = binding.get_mut(&key).ok_or(SystemError::EINVAL)?; 365 366 // 确保后面的唤醒操作是有意义的 367 if bucket_mut.chain.is_empty() { 368 return Ok(0); 369 } 370 // 从队列中唤醒 371 let count = bucket_mut.wake_up(key.clone(), Some(bitset), nr_wake)?; 372 373 drop(binding); 374 375 FutexData::try_remove(&key); 376 377 Ok(count) 378 } 379 380 /// ### 唤醒制定uaddr1上的最多nr_wake个进程，然后将uaddr1最多nr_requeue个进程移动到uaddr2绑定的futex上 futex_requeue( uaddr1: VirtAddr, flags: FutexFlag, uaddr2: VirtAddr, nr_wake: i32, nr_requeue: i32, cmpval: Option<u32>, requeue_pi: bool, ) -> Result<usize, SystemError>381 pub fn futex_requeue( 382 uaddr1: VirtAddr, 383 flags: FutexFlag, 384 uaddr2: VirtAddr, 385 nr_wake: i32, 386 nr_requeue: i32, 387 cmpval: Option<u32>, 388 requeue_pi: bool, 389 ) -> Result<usize, SystemError> { 390 if nr_requeue < 0 || nr_wake < 0 { 391 return Err(SystemError::EINVAL); 392 } 393 394 // 暂时不支持优先级继承 395 if requeue_pi { 396 return Err(SystemError::ENOSYS); 397 } 398 399 let key1 = Self::get_futex_key( 400 uaddr1, 401 flags.contains(FutexFlag::FLAGS_SHARED), 402 FutexAccess::FutexRead, 403 )?; 404 let key2 = Self::get_futex_key(uaddr2, flags.contains(FutexFlag::FLAGS_SHARED), { 405 match requeue_pi { 406 true => FutexAccess::FutexWrite, 407 false => FutexAccess::FutexRead, 408 } 409 })?; 410 411 if requeue_pi && key1 == key2 { 412 return Err(SystemError::EINVAL); 413 } 414 415 if likely(!cmpval.is_none()) { 416 let uval_reader = 417 UserBufferReader::new(uaddr1.as_ptr::<u32>(), core::mem::size_of::<u32>(), true)?; 418 let curval = uval_reader.read_one_from_user::<u32>(0)?; 419 420 // 判断是否满足条件 421 if *curval != cmpval.unwrap() { 422 return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); 423 } 424 } 425 426 let mut futex_data_guard = FutexData::futex_map(); 427 if !requeue_pi { 428 // 唤醒nr_wake个进程 429 let bucket_1_mut = futex_data_guard.get_mut(&key1).ok_or(SystemError::EINVAL)?; 430 let ret = bucket_1_mut.wake_up(key1.clone(), None, nr_wake as u32)?; 431 // 将bucket1中最多nr_requeue个任务转移到bucket2 432 for _ in 0..nr_requeue { 433 let bucket_1_mut = futex_data_guard.get_mut(&key1).ok_or(SystemError::EINVAL)?; 434 let futex_q = bucket_1_mut.chain.pop_front(); 435 match futex_q { 436 Some(futex_q) => { 437 let bucket_2_mut = 438 futex_data_guard.get_mut(&key2).ok_or(SystemError::EINVAL)?; 439 bucket_2_mut.chain.push_back(futex_q); 440 } 441 None => { 442 break; 443 } 444 } 445 } 446 447 return Ok(ret); 448 } else { 449 // 暂时不支持优先级继承 450 todo!() 451 } 452 } 453 454 /// ### 唤醒futex上的进程的同时进行一些操作 futex_wake_op( uaddr1: VirtAddr, flags: FutexFlag, uaddr2: VirtAddr, nr_wake: i32, nr_wake2: i32, op: i32, ) -> Result<usize, SystemError>455 pub fn futex_wake_op( 456 uaddr1: VirtAddr, 457 flags: FutexFlag, 458 uaddr2: VirtAddr, 459 nr_wake: i32, 460 nr_wake2: i32, 461 op: i32, 462 ) -> Result<usize, SystemError> { 463 let key1 = Futex::get_futex_key( 464 uaddr1, 465 flags.contains(FutexFlag::FLAGS_SHARED), 466 FutexAccess::FutexRead, 467 )?; 468 let key2 = Futex::get_futex_key( 469 uaddr2, 470 flags.contains(FutexFlag::FLAGS_SHARED), 471 FutexAccess::FutexWrite, 472 )?; 473 474 let mut futex_data_guard = FutexData::futex_map(); 475 let bucket1 = futex_data_guard.get_mut(&key1).ok_or(SystemError::EINVAL)?; 476 let mut wake_count = 0; 477 478 // 唤醒uaddr1中的进程 479 wake_count += bucket1.wake_up(key1, None, nr_wake as u32)?; 480 481 match Self::futex_atomic_op_inuser(op as u32, uaddr2) { 482 Ok(ret) => { 483 // 操作成功则唤醒uaddr2中的进程 484 if ret { 485 let bucket2 = futex_data_guard.get_mut(&key2).ok_or(SystemError::EINVAL)?; 486 wake_count += bucket2.wake_up(key2, None, nr_wake2 as u32)?; 487 } 488 } 489 Err(e) => { 490 // TODO:retry? 491 return Err(e); 492 } 493 } 494 495 Ok(wake_count) 496 } 497 get_futex_key( uaddr: VirtAddr, fshared: bool, _access: FutexAccess, ) -> Result<FutexKey, SystemError>498 fn get_futex_key( 499 uaddr: VirtAddr, 500 fshared: bool, 501 _access: FutexAccess, 502 ) -> Result<FutexKey, SystemError> { 503 let mut address = uaddr.data(); 504 505 // 计算相对页的偏移量 506 let offset = address & (MMArch::PAGE_SIZE - 1); 507 // 判断内存对齐 508 if !(uaddr.data() & (core::mem::size_of::<u32>() - 1) == 0) { 509 return Err(SystemError::EINVAL); 510 } 511 512 // 目前address指向所在页面的起始地址 513 address -= offset; 514 515 // 若不是进程间共享的futex，则返回Private 516 if !fshared { 517 return Ok(FutexKey { 518 ptr: 0, 519 word: 0, 520 offset: offset as u32, 521 key: InnerFutexKey::Private(PrivateKey { 522 address: address as u64, 523 address_space: None, 524 }), 525 }); 526 } 527 528 // 获取到地址所在地址空间 529 let address_space = AddressSpace::current()?; 530 // TODO： 判断是否为匿名映射，是匿名映射才返回PrivateKey 531 return Ok(FutexKey { 532 ptr: 0, 533 word: 0, 534 offset: offset as u32, 535 key: InnerFutexKey::Private(PrivateKey { 536 address: address as u64, 537 address_space: Some(Arc::downgrade(&address_space)), 538 }), 539 }); 540 541 // 未实现共享内存机制,贡献内存部分应该通过inode构建SharedKey 542 // todo!("Shared memory not implemented"); 543 } 544 futex_atomic_op_inuser(encoded_op: u32, uaddr: VirtAddr) -> Result<bool, SystemError>545 pub fn futex_atomic_op_inuser(encoded_op: u32, uaddr: VirtAddr) -> Result<bool, SystemError> { 546 let op = FutexOP::from_bits((encoded_op & 0x70000000) >> 28).ok_or(SystemError::ENOSYS)?; 547 let cmp = 548 FutexOpCMP::from_bits((encoded_op & 0x0f000000) >> 24).ok_or(SystemError::ENOSYS)?; 549 550 let sign_extend32 = |value: u32, index: i32| { 551 let shift = (31 - index) as u8; 552 return (value << shift) >> shift; 553 }; 554 555 let mut oparg = sign_extend32((encoded_op & 0x00fff000) >> 12, 11); 556 let cmparg = sign_extend32(encoded_op & 0x00000fff, 11); 557 558 if encoded_op & (FutexOP::FUTEX_OP_OPARG_SHIFT.bits() << 28) != 0 { 559 if oparg > 31 { 560 kwarn!( 561 "futex_wake_op: pid:{} tries to shift op by {}; fix this program", 562 ProcessManager::current_pcb().pid().data(), 563 oparg 564 ); 565 566 oparg &= 31; 567 } 568 } 569 570 // TODO: 这个汇编似乎是有问题的，目前不好测试 571 let old_val = Self::arch_futex_atomic_op_inuser(op, oparg, uaddr)?; 572 573 match cmp { 574 FutexOpCMP::FUTEX_OP_CMP_EQ => { 575 return Ok(cmparg == old_val); 576 } 577 FutexOpCMP::FUTEX_OP_CMP_NE => { 578 return Ok(cmparg != old_val); 579 } 580 FutexOpCMP::FUTEX_OP_CMP_LT => { 581 return Ok(cmparg < old_val); 582 } 583 FutexOpCMP::FUTEX_OP_CMP_LE => { 584 return Ok(cmparg <= old_val); 585 } 586 FutexOpCMP::FUTEX_OP_CMP_GE => { 587 return Ok(cmparg >= old_val); 588 } 589 FutexOpCMP::FUTEX_OP_CMP_GT => { 590 return Ok(cmparg > old_val); 591 } 592 _ => { 593 return Err(SystemError::ENOSYS); 594 } 595 } 596 } 597 598 /// ### 对futex进行操作 599 /// 600 /// 进入该方法会关闭中断保证修改的原子性，所以进入该方法前应确保中断锁已释放 601 /// 602 /// ### return uaddr原来的值 603 #[allow(unused_assignments)] arch_futex_atomic_op_inuser( op: FutexOP, oparg: u32, uaddr: VirtAddr, ) -> Result<u32, SystemError>604 pub fn arch_futex_atomic_op_inuser( 605 op: FutexOP, 606 oparg: u32, 607 uaddr: VirtAddr, 608 ) -> Result<u32, SystemError> { 609 let guard = unsafe { CurrentIrqArch::save_and_disable_irq() }; 610 611 let reader = 612 UserBufferReader::new(uaddr.as_ptr::<u32>(), core::mem::size_of::<u32>(), true)?; 613 614 let oldval = reader.read_one_from_user::<u32>(0)?; 615 616 let atomic_addr = AtomicU64::new(uaddr.data() as u64); 617 // 这个指针是指向指针的指针 618 let ptr = atomic_addr.as_ptr(); 619 match op { 620 FutexOP::FUTEX_OP_SET => unsafe { 621 *((*ptr) as *mut u32) = oparg; 622 }, 623 FutexOP::FUTEX_OP_ADD => unsafe { 624 *((*ptr) as *mut u32) += oparg; 625 }, 626 FutexOP::FUTEX_OP_OR => unsafe { 627 *((*ptr) as *mut u32) |= oparg; 628 }, 629 FutexOP::FUTEX_OP_ANDN => unsafe { 630 *((*ptr) as *mut u32) &= oparg; 631 }, 632 FutexOP::FUTEX_OP_XOR => unsafe { 633 *((*ptr) as *mut u32) ^= oparg; 634 }, 635 _ => return Err(SystemError::ENOSYS), 636 } 637 638 drop(guard); 639 640 Ok(*oldval) 641 } 642 } 643