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 9 use crate::{ 10 arch::{sched::sched, CurrentIrqArch, MMArch}, 11 exception::InterruptArch, 12 libs::spinlock::{SpinLock, SpinLockGuard}, 13 mm::{ucontext::AddressSpace, MemoryManagementArch, VirtAddr}, 14 process::{ProcessControlBlock, ProcessManager}, 15 syscall::{user_access::UserBufferReader, SystemError}, 16 time::{ 17 timer::{next_n_us_timer_jiffies, Timer, WakeUpHelper}, 18 TimeSpec, 19 }, 20 }; 21 22 use super::constant::*; 23 24 static mut FUTEX_DATA: Option<FutexData> = None; 25 26 pub struct FutexData { 27 data: SpinLock<HashMap<FutexKey, FutexHashBucket>>, 28 } 29 30 impl FutexData { 31 pub fn futex_map() -> SpinLockGuard<'static, HashMap<FutexKey, FutexHashBucket>> { 32 unsafe { FUTEX_DATA.as_ref().unwrap().data.lock() } 33 } 34 35 pub fn try_remove(key: &FutexKey) -> Option<FutexHashBucket> { 36 unsafe { 37 let mut guard = FUTEX_DATA.as_ref().unwrap().data.lock(); 38 if let Some(futex) = guard.get(key) { 39 if futex.chain.is_empty() { 40 return guard.remove(key); 41 } 42 } 43 } 44 None 45 } 46 } 47 48 pub struct Futex; 49 50 // 对于同一个futex的进程或线程将会在这个bucket等待 51 pub struct FutexHashBucket { 52 // 该futex维护的等待队列 53 chain: LinkedList<Arc<FutexObj>>, 54 } 55 56 impl FutexHashBucket { 57 /// ## 判断是否在bucket里 58 pub fn contains(&self, futex_q: &FutexObj) -> bool { 59 self.chain 60 .iter() 61 .filter(|x| futex_q.pcb.ptr_eq(&x.pcb) && x.key == futex_q.key) 62 .count() 63 != 0 64 } 65 66 /// 让futex_q在该bucket上挂起 67 /// 68 /// 进入该函数前，需要关中断 69 #[inline(always)] 70 pub fn sleep_no_sched(&mut self, futex_q: Arc<FutexObj>) -> Result<(), SystemError> { 71 assert!(CurrentIrqArch::is_irq_enabled() == false); 72 self.chain.push_back(futex_q); 73 74 ProcessManager::mark_sleep(true)?; 75 76 Ok(()) 77 } 78 79 /// ## 唤醒队列中的最多nr_wake个进程 80 /// 81 /// return: 唤醒的进程数 82 #[inline(always)] 83 pub fn wake_up( 84 &mut self, 85 key: FutexKey, 86 bitset: Option<u32>, 87 nr_wake: u32, 88 ) -> Result<usize, SystemError> { 89 let mut count = 0; 90 let mut pop_count = 0; 91 while let Some(futex_q) = self.chain.pop_front() { 92 if futex_q.key == key { 93 // TODO: 考虑优先级继承的机制 94 95 if let Some(bitset) = bitset { 96 if futex_q.bitset != bitset { 97 self.chain.push_back(futex_q); 98 continue; 99 } 100 } 101 102 // 唤醒 103 if futex_q.pcb.upgrade().is_some() { 104 self.remove(futex_q.clone()); 105 ProcessManager::wakeup(&futex_q.pcb.upgrade().unwrap())?; 106 } 107 108 // 判断唤醒数 109 count += 1; 110 if count >= nr_wake { 111 break; 112 } 113 } else { 114 self.chain.push_back(futex_q); 115 } 116 // 判断是否循环完队列了 117 pop_count += 1; 118 if pop_count >= self.chain.len() { 119 break; 120 } 121 } 122 Ok(count as usize) 123 } 124 125 /// 将FutexObj从bucket中删除 126 pub fn remove(&mut self, futex: Arc<FutexObj>) { 127 self.chain.drain_filter(|x| Arc::ptr_eq(x, &futex)); 128 } 129 } 130 131 #[derive(Debug)] 132 pub struct FutexObj { 133 pcb: Weak<ProcessControlBlock>, 134 key: FutexKey, 135 bitset: u32, 136 // TODO: 优先级继承 137 } 138 139 pub enum FutexAccess { 140 FutexRead, 141 FutexWrite, 142 } 143 144 #[allow(dead_code)] 145 #[derive(Hash, PartialEq, Eq, Clone, Debug)] 146 /// ### 用于定位内核唯一的futex 147 pub enum InnerFutexKey { 148 Shared(SharedKey), 149 Private(PrivateKey), 150 } 151 152 #[derive(Hash, PartialEq, Eq, Clone, Debug)] 153 pub struct FutexKey { 154 ptr: u64, 155 word: u64, 156 offset: u32, 157 key: InnerFutexKey, 158 } 159 160 /// 不同进程间通过文件共享futex变量，表明该变量在文件中的位置 161 #[derive(Hash, PartialEq, Eq, Clone, Debug)] 162 pub struct SharedKey { 163 i_seq: u64, 164 page_offset: u64, 165 } 166 167 /// 同一进程的不同线程共享futex变量，表明该变量在进程地址空间中的位置 168 #[derive(Clone, Debug)] 169 pub struct PrivateKey { 170 // 所在的地址空间 171 address_space: Option<Weak<AddressSpace>>, 172 // 表示所在页面的初始地址 173 address: u64, 174 } 175 176 impl Hash for PrivateKey { 177 fn hash<H: ~const Hasher>(&self, state: &mut H) { 178 self.address.hash(state); 179 } 180 } 181 182 impl Eq for PrivateKey {} 183 184 impl PartialEq for PrivateKey { 185 fn eq(&self, other: &Self) -> bool { 186 if self.address_space.is_none() && other.address_space.is_none() { 187 return self.address == other.address; 188 } else { 189 return self 190 .address_space 191 .as_ref() 192 .unwrap_or(&Weak::default()) 193 .ptr_eq(&other.address_space.as_ref().unwrap_or(&Weak::default())) 194 && self.address == other.address; 195 } 196 } 197 } 198 199 impl Futex { 200 /// ### 初始化FUTEX_DATA 201 pub fn init() { 202 unsafe { 203 FUTEX_DATA = Some(FutexData { 204 data: SpinLock::new(HashMap::new()), 205 }) 206 }; 207 } 208 209 /// ### 让当前进程在指定futex上等待直到futex_wake显式唤醒 210 pub fn futex_wait( 211 uaddr: VirtAddr, 212 flags: FutexFlag, 213 val: u32, 214 abs_time: Option<TimeSpec>, 215 bitset: u32, 216 ) -> Result<usize, SystemError> { 217 if bitset == 0 { 218 return Err(SystemError::EINVAL); 219 } 220 221 // 获取全局hash表的key值 222 let key = Self::get_futex_key( 223 uaddr, 224 flags.contains(FutexFlag::FLAGS_SHARED), 225 FutexAccess::FutexRead, 226 )?; 227 228 let mut futex_map_guard = FutexData::futex_map(); 229 let bucket = futex_map_guard.get_mut(&key); 230 let bucket_mut = match bucket { 231 Some(bucket) => bucket, 232 None => { 233 let bucket = FutexHashBucket { 234 chain: LinkedList::new(), 235 }; 236 futex_map_guard.insert(key.clone(), bucket); 237 futex_map_guard.get_mut(&key).unwrap() 238 } 239 }; 240 241 // 使用UserBuffer读取futex 242 let user_reader = 243 UserBufferReader::new(uaddr.as_ptr::<u32>(), core::mem::size_of::<u32>(), true)?; 244 245 // 从用户空间读取到futex的val 246 let mut uval = 0; 247 248 // 读取 249 // 这里只尝试一种方式去读取用户空间，与linux不太一致 250 // 对于linux，如果bucket被锁住时读取失败，将会将bucket解锁后重新读取 251 user_reader.copy_one_from_user::<u32>(&mut uval, 0)?; 252 253 // 不满足wait条件，返回错误 254 if uval != val { 255 return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); 256 } 257 258 let pcb = ProcessManager::current_pcb(); 259 // 创建超时计时器任务 260 let mut timer = None; 261 if !abs_time.is_none() { 262 let time = abs_time.unwrap(); 263 264 let wakeup_helper = WakeUpHelper::new(pcb.clone()); 265 266 let sec = time.tv_sec; 267 let nsec = time.tv_nsec; 268 let jiffies = next_n_us_timer_jiffies((nsec / 1000 + sec * 1_000_000) as u64); 269 270 let wake_up = Timer::new(wakeup_helper, jiffies); 271 272 wake_up.activate(); 273 timer = Some(wake_up); 274 } 275 276 let futex_q = Arc::new(FutexObj { 277 pcb: Arc::downgrade(&pcb), 278 key: key.clone(), 279 bitset, 280 }); 281 let irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() }; 282 // 满足条件则将当前进程在该bucket上挂起 283 bucket_mut.sleep_no_sched(futex_q.clone()).map_err(|e| { 284 kwarn!("error:{e:?}"); 285 e 286 })?; 287 drop(bucket_mut); 288 drop(futex_map_guard); 289 drop(irq_guard); 290 sched(); 291 292 // 被唤醒后的检查 293 let mut futex_map_guard = FutexData::futex_map(); 294 let bucket = futex_map_guard.get_mut(&key); 295 let bucket_mut = match bucket { 296 // 如果该pcb不在链表里面了或者该链表已经被释放，就证明是正常的Wake操作 297 Some(bucket_mut) => { 298 if !bucket_mut.contains(&futex_q) { 299 // 取消定时器任务 300 if timer.is_some() { 301 timer.unwrap().cancel(); 302 } 303 return Ok(0); 304 } 305 // 非正常唤醒，返回交给下层 306 bucket_mut 307 } 308 None => { 309 // 取消定时器任务 310 if timer.is_some() { 311 timer.unwrap().cancel(); 312 } 313 return Ok(0); 314 } 315 }; 316 317 // 如果是超时唤醒，则返回错误 318 if timer.is_some() { 319 if timer.clone().unwrap().timeout() { 320 bucket_mut.remove(futex_q); 321 322 return Err(SystemError::ETIMEDOUT); 323 } 324 } 325 326 // TODO: 如果没有挂起的信号，则重新判断是否满足wait要求，重新进入wait 327 328 // 经过前面的几个判断，到这里之后， 329 // 当前进程被唤醒大概率是其他进程更改了uval,需要重新去判断当前进程是否满足wait 330 331 // 到这里之后，前面的唤醒条件都不满足，则是被信号唤醒 332 // 需要处理信号然后重启futex系统调用 333 334 // 取消定时器任务 335 if timer.is_some() { 336 let timer = timer.unwrap(); 337 if !timer.timeout() { 338 timer.cancel(); 339 } 340 } 341 Ok(0) 342 } 343 344 // ### 唤醒指定futex上挂起的最多nr_wake个进程 345 pub fn futex_wake( 346 uaddr: VirtAddr, 347 flags: FutexFlag, 348 nr_wake: u32, 349 bitset: u32, 350 ) -> Result<usize, SystemError> { 351 if bitset == 0 { 352 return Err(SystemError::EINVAL); 353 } 354 355 // 获取futex_key,并且判断地址空间合法性 356 let key = Self::get_futex_key( 357 uaddr, 358 flags.contains(FutexFlag::FLAGS_SHARED), 359 FutexAccess::FutexRead, 360 )?; 361 let mut binding = FutexData::futex_map(); 362 let bucket_mut = binding.get_mut(&key).ok_or(SystemError::EINVAL)?; 363 364 // 确保后面的唤醒操作是有意义的 365 if bucket_mut.chain.is_empty() { 366 return Ok(0); 367 } 368 // 从队列中唤醒 369 let count = bucket_mut.wake_up(key.clone(), Some(bitset), nr_wake)?; 370 371 drop(bucket_mut); 372 drop(binding); 373 374 FutexData::try_remove(&key); 375 376 Ok(count) 377 } 378 379 /// ### 唤醒制定uaddr1上的最多nr_wake个进程，然后将uaddr1最多nr_requeue个进程移动到uaddr2绑定的futex上 380 pub fn futex_requeue( 381 uaddr1: VirtAddr, 382 flags: FutexFlag, 383 uaddr2: VirtAddr, 384 nr_wake: i32, 385 nr_requeue: i32, 386 cmpval: Option<u32>, 387 requeue_pi: bool, 388 ) -> Result<usize, SystemError> { 389 if nr_requeue < 0 || nr_wake < 0 { 390 return Err(SystemError::EINVAL); 391 } 392 393 // 暂时不支持优先级继承 394 if requeue_pi { 395 return Err(SystemError::ENOSYS); 396 } 397 398 let key1 = Self::get_futex_key( 399 uaddr1, 400 flags.contains(FutexFlag::FLAGS_SHARED), 401 FutexAccess::FutexRead, 402 )?; 403 let key2 = Self::get_futex_key(uaddr2, flags.contains(FutexFlag::FLAGS_SHARED), { 404 match requeue_pi { 405 true => FutexAccess::FutexWrite, 406 false => FutexAccess::FutexRead, 407 } 408 })?; 409 410 if requeue_pi && key1 == key2 { 411 return Err(SystemError::EINVAL); 412 } 413 414 if likely(!cmpval.is_none()) { 415 let uval_reader = 416 UserBufferReader::new(uaddr1.as_ptr::<u32>(), core::mem::size_of::<u32>(), true)?; 417 let curval = uval_reader.read_one_from_user::<u32>(0)?; 418 419 // 判断是否满足条件 420 if *curval != cmpval.unwrap() { 421 return Err(SystemError::EAGAIN_OR_EWOULDBLOCK); 422 } 423 } 424 425 let mut futex_data_guard = FutexData::futex_map(); 426 if !requeue_pi { 427 // 唤醒nr_wake个进程 428 let bucket_1_mut = futex_data_guard.get_mut(&key1).ok_or(SystemError::EINVAL)?; 429 let ret = bucket_1_mut.wake_up(key1.clone(), None, nr_wake as u32)?; 430 drop(bucket_1_mut); 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上的进程的同时进行一些操作 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 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 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)] 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_mut_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 644 #[no_mangle] 645 unsafe extern "C" fn rs_futex_init() { 646 Futex::init(); 647 } 648