1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

use futures::{Future, Poll, Async};
use core::{OclPrm, MemMap as MemMapCore, Mem};
use standard::{Event, Queue, EventList};
use super::{Error as AsyncError, Result as AsyncResult, MemMap};


/// A future which resolves to a `MemMap` as soon as its creating command
/// completes.
///
/// [UNSTABLE]: This types methods may be renamed or otherwise changed at any time.
#[must_use = "futures do nothing unless polled"]
#[derive(Debug)]
pub struct FutureMemMap<T: OclPrm> {
    core: Option<MemMapCore<T>>,
    len: usize,
    map_event: Event,
    unmap_wait_list: Option<EventList>,
    unmap_target_event: Option<Event>,
    buffer: Option<Mem>,
    queue: Option<Queue>,
    callback_is_set: bool,
}

impl<T: OclPrm> FutureMemMap<T> {
    /// Returns a new `FutureMemMap`.
    pub unsafe fn new(core: MemMapCore<T>, len: usize, map_event: Event, buffer: Mem, queue: Queue)
            -> FutureMemMap<T>
    {
        FutureMemMap {
            core: Some(core),
            len: len,
            map_event: map_event,
            unmap_wait_list: None,
            unmap_target_event: None,
            buffer: Some(buffer),
            queue: Some(queue),
            callback_is_set: false,
        }
    }

    /// Set an event wait list for the unmap command.
    ///
    /// Setting a wait list here will disallow any wait list from being set
    /// later if/when calling unmap manually.
    ///
    /// [UNSTABLE]: This method may be renamed or otherwise changed.
    pub fn set_unmap_wait_list<El>(&mut self, wait_list: El) where El: Into<EventList> {
        self.unmap_wait_list = Some(wait_list.into())
    }

    /// Create an event which will be triggered (set complete) after this
    /// future resolves into a `MemMap` **and** after that `MemMap` is dropped
    /// or manually unmapped.
    ///
    /// The returned event can be added to the wait list of subsequent OpenCL
    /// commands with the expectation that when all preceeding futures are
    /// complete, the event will automatically be 'triggered' by having its
    /// status set to complete, causing those commands to execute. This can be
    /// used to inject host side code in amongst OpenCL commands without
    /// thread blocking or extra delays of any kind.
    ///
    /// [UNSTABLE]: This method may be renamed or otherwise changed.
    pub fn create_unmap_target_event(&mut self) -> AsyncResult<&mut Event> {
        if let Some(ref queue) = self.queue {
            let uev = Event::user(&queue.context())?;
            self.unmap_target_event = Some(uev);
            Ok(self.unmap_target_event.as_mut().unwrap())
        } else {
            Err("FutureMemMap::create_unmap_target_event: No queue found!".into())
        }
    }

    /// Blocks the current thread until the OpenCL command is complete and an
    /// appropriate lock can be obtained on the underlying data.
    pub fn wait(self) -> AsyncResult<MemMap<T>> {
        <Self as Future>::wait(self)
    }

    /// Returns the unmap event if it has been created.
    ///
    /// [UNSTABLE]: This method may be renamed or otherwise changed.
    #[inline]
    pub fn unmap_target_event(&self) -> Option<&Event> {
        self.unmap_target_event.as_ref()
    }

    /// Resolves this `FutureMemMap` into a `MemMap`.
    fn to_mapped_mem(&mut self) -> AsyncResult<MemMap<T>> {
        let joined = self.core.take().and_then(|core| {
            self.buffer.take().and_then(|buf| {
                self.queue.take().and_then(|queue| {
                    Some((core, buf, queue))
                })
            })
        });

        match joined {
            Some((core, buffer, queue)) => {
                unsafe { Ok(MemMap::new(core, self.len, self.unmap_wait_list.take(),
                    self.unmap_target_event.take(), buffer, queue )) }
            },
            _ => Err("FutureMemMap::create_unmap_target_event: No queue and/or buffer found!".into()),
        }
    }
}

#[cfg(not(feature = "async_block"))]
impl<T> Future for FutureMemMap<T> where T: OclPrm + 'static {
    type Item = MemMap<T>;
    type Error = AsyncError;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        // println!("Polling FutureMemMap...");
        match self.map_event.is_complete() {
            Ok(true) => {
                self.to_mapped_mem().map(|mm| Async::Ready(mm))
            }
            Ok(false) => {
                if !self.callback_is_set {
                    self.map_event.set_unpark_callback()?;
                    self.callback_is_set = true;
                }

                Ok(Async::NotReady)
            },
            Err(err) => Err(err.into()),
        }
    }
}

/// Blocking implementation.
#[cfg(feature = "async_block")]
impl<T: OclPrm> Future for FutureMemMap<T> {
    type Item = MemMap<T>;
    type Error = AsyncError;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        // println!("Polling FutureMemMap...");
        let _ = self.callback_is_set;
        self.map_event.wait_for()?;
        self.to_mapped_mem().map(|mm| Async::Ready(mm))
    }
}

unsafe impl<T: OclPrm> Send for FutureMemMap<T> {}
unsafe impl<T: OclPrm> Sync for FutureMemMap<T> {}