OC bottom principle exploration of GCD source analysis final chapter

This is the seventh day of my participation in the August More text Challenge. For details, see:August is more challenging

Barrier function

The most direct effect of the fence function is to control the order of execution of tasks to produce synchronization effect.

• dispatch_barrier_async: I will come here after the previous task is completed
• dispatch_barrier_sync: has the same effect, but it blocks the thread and affects the execution of subsequent functions.

Example demonstrates

- (void)demo1 {
    dispatch_queue_t concurrentQueue = dispatch_queue_create("cooci", DISPATCH_QUEUE_CONCURRENT);
    // This is ok!
    /* 1. Asynchronous function */
    dispatch_async(concurrentQueue, ^{
        NSLog(@ "123");
    });
    /* 2. Asynchronous function */
    dispatch_async(concurrentQueue, ^{
        sleep(1);
        NSLog(@ "456");
    });
    /* 3. */ // - dispatch_barrier_sync
    dispatch_barrier_async(concurrentQueue, ^{
        NSLog(@ "-- % @ -- -- -- -- --"[NSThread currentThread]);
    });
    /* 4. Asynchronous function */
    dispatch_async(concurrentQueue, ^{
        NSLog(@" Loaded so much, take a breath!!");
    });
    / / 5
    NSLog(@"********** get up and do it!!");
}
Copy the code

The asynchronous concurrent queue is used here, and the fence function is used when the asynchronous concurrent is used. The previous task will come here after the completion of the execution, but it will not block the execution of the later task, so the fence function in Step 3 must be executed after Steps 1 and 2. Step 1 and step 2 are not ordered, and Step 4 and Step 5 are not ordered. Note: What if we replace this concurrent queue with a global concurrent queue?

dispatch_queue_t concurrentQueue = dispatch_get_global_queue(0.0);
Copy the code

What we find is that the fence function doesn’t work anymore,That is, the fence function in the concurrent queue is invalid in the global concurrent queueThen why? As usual, we need the source code from the previous Dispatch to get a peek

dispatch_barrier_sync

void
dispatch_barrier_sync(dispatch_queue_t dq, dispatch_block_t work)
{
	uintptr_t dc_flags = DC_FLAG_BARRIER | DC_FLAG_BLOCK;
	if (unlikely(_dispatch_block_has_private_data(work))) {
		return _dispatch_sync_block_with_privdata(dq, work, dc_flags);
	}
	_dispatch_barrier_sync_f(dq, work, _dispatch_Block_invoke(work), dc_flags);
}
Copy the code

So this code is similar to the one I introduced for the synchronization function, We locate the call chain “dispatch_barrier_sync” -> “_dispatch_barrier_sync_f_inline” we locate the call chain “dispatch_barrier_sync” -> “_dispatch_barrier_sync_f_inline” -> _dispatch_sync_invoke_and_complete_recurse -> _dispatch_sync_complete_recurse

static void
_dispatch_sync_complete_recurse(dispatch_queue_t dq, dispatch_queue_t stop_dq,
		uintptr_t dc_flags)
{
	bool barrier = (dc_flags & DC_FLAG_BARRIER);
	do {
		if (dq == stop_dq) return;
        // If there is a barrier, all queues in front of it will be executed
		if (barrier) {
			dx_wakeup(dq, 0, DISPATCH_WAKEUP_BARRIER_COMPLETE);
		} else {
            // There is no common synchronization function to perform
			_dispatch_lane_non_barrier_complete(upcast(dq)._dl, 0);
		}
		dq = dq->do_targetq;
		barrier = (dq->dq_width == 1);
	} while (unlikely(dq->do_targetq));
}
Copy the code

Dx_wakeup if there is a barrier function

#define dx_wakeup(x, y, z) dx_vtable(x)->dq_wakeup(x, y, z)
Copy the code

DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_concurrent, lane,
	.do_type        = DISPATCH_QUEUE_CONCURRENT_TYPE,
	.do_dispose     = _dispatch_lane_dispose,
	.do_debug       = _dispatch_queue_debug,
	.do_invoke      = _dispatch_lane_invoke,

	.dq_activate    = _dispatch_lane_activate,
	.dq_wakeup      = _dispatch_lane_wakeup,
	.dq_push        = _dispatch_lane_concurrent_push,
);

DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_global, lane,
	.do_type        = DISPATCH_QUEUE_GLOBAL_ROOT_TYPE,
	.do_dispose     = _dispatch_object_no_dispose,
	.do_debug       = _dispatch_queue_debug,
	.do_invoke      = _dispatch_object_no_invoke,

	.dq_activate    = _dispatch_queue_no_activate,
	.dq_wakeup      = _dispatch_root_queue_wakeup,
	.dq_push        = _dispatch_root_queue_push,
);
Copy the code

=_dispatch_lane_wakeup for self-created concurrent queues and =_dispatch_root_queue_wakeup for global concurrent queues

queue_concurrent VS queue_global

void
_dispatch_lane_wakeup(dispatch_lane_class_t dqu, dispatch_qos_t qos,
		dispatch_wakeup_flags_t flags)
{
	dispatch_queue_wakeup_target_t target = DISPATCH_QUEUE_WAKEUP_NONE;

	if (unlikely(flags & DISPATCH_WAKEUP_BARRIER_COMPLETE)) {
		return _dispatch_lane_barrier_complete(dqu, qos, flags);
	}
	if (_dispatch_queue_class_probe(dqu)) {
		target = DISPATCH_QUEUE_WAKEUP_TARGET;
	}
	return _dispatch_queue_wakeup(dqu, qos, flags, target);
}
Copy the code

void
_dispatch_root_queue_wakeup(dispatch_queue_global_t dq,
		DISPATCH_UNUSED dispatch_qos_t qos, dispatch_wakeup_flags_t flags)
{
	if(! (flags & DISPATCH_WAKEUP_BLOCK_WAIT)) { DISPATCH_INTERNAL_CRASH(dq->dq_priority,"Don't try to wake up or override a root queue");
	}
	if (flags & DISPATCH_WAKEUP_CONSUME_2) {
		return_dispatch_release_2_tailcall(dq); }}Copy the code

The differences are also evident in the source code. There is no barrier function in the global concurrent queue, but there is a barrier function judgment _dispatch_lane_barrier_complete in the self-created concurrent queue

static void
_dispatch_lane_barrier_complete(dispatch_lane_class_t dqu, dispatch_qos_t qos,
		dispatch_wakeup_flags_t flags)
{
	dispatch_queue_wakeup_target_t target = DISPATCH_QUEUE_WAKEUP_NONE;
	dispatch_lane_t dq = dqu._dl;

	if(dq->dq_items_tail && ! DISPATCH_QUEUE_IS_SUSPENDED(dq)) {struct dispatch_object_s *dc = _dispatch_queue_get_head(dq);
        // The synchronization function
		if (likely(dq->dq_width == 1 || _dispatch_object_is_barrier(dc))) {
			if (_dispatch_object_is_waiter(dc)) {
				return _dispatch_lane_drain_barrier_waiter(dq, dc, flags, 0); }}else if (dq->dq_width > 1 && !_dispatch_object_is_barrier(dc)) {
			return _dispatch_lane_drain_non_barriers(dq, dc, flags);
		}
		// ...
	}
	/ /...
	return _dispatch_lane_class_barrier_complete(dq, qos, flags, target, owned);
}
Copy the code

If it’s synchronous it’s going to wait otherwise it’s going to complete _dispatch_LANe_class_barrier_complete, so make sure that all the previous tasks are done.

This also verifies the above conclusion: the global concurrent queue does not deal with the fence function correlation, so the fence function is useless in the global concurrent queue. The reason for this is that system-level calls to global concurrent queues are also made, and the fence function essentially blocks the current thread, thus affecting efficiency. The fence function must be used in the same queue. For example, when using AFN, we can not get the current queue of AFN. Therefore, the fence function is usually used in few scenarios, but the most used is scheduling group

A semaphoredispatch_semaphore_t

Dispatch_semaphore_create Creates a semaphore. The number in this number is the maximum number of concurrent requests. Dispatch_semaphore_wait Wait -1

- (void)test {	
	dispatch_queue_t queue = dispatch_get_global_queue(0.0);
   dispatch_semaphore_t sem = dispatch_semaphore_create(0); 
    1 / / task
    dispatch_async(queue, ^{
        dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER); / / wait for
        NSLog(@" Execute mission 1");
        NSLog(@" Mission 1 completed");
    });
    
    2 / / task
    dispatch_async(queue, ^{
        sleep(2);
        NSLog(@" Mission 2");
        NSLog(@" Mission 2 completed");
        dispatch_semaphore_signal(sem); // Send signal +1
    });
    
    3 / / task
    dispatch_async(queue, ^{
        dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
        sleep(2);
        NSLog(@" Mission 3");
        NSLog(@" Mission 3 completed");
        dispatch_semaphore_signal(sem);
    });
 }
Copy the code

Mission 3 will not go ahead. It will wait forever.

dispatch_semaphore_create

 * @param value
 * The starting value for the semaphore. Passing a value less than zero will
 * cause NULL to be returned.
Copy the code

The starting value of the semaphore. Passing a value less than zero will result in a return of NULL.

dispatch_semaphore_signal

intptr_t
dispatch_semaphore_signal(dispatch_semaphore_t dsema) / / 0
{
	long value = os_atomic_inc2o(dsema, dsema_value, release); // +=1 value = 1
	if (likely(value > 0)) {
		return 0;   // Return 0 directly
	}
	if (unlikely(value == LONG_MIN)) {
		DISPATCH_CLIENT_CRASH(value,
				"Unbalanced call to dispatch_semaphore_signal()");
	}
	return _dispatch_semaphore_signal_slow(dsema);
}
Copy the code

dispatch_semaphore_wait

intptr_t
dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
	long value = os_atomic_dec2o(dsema, dsema_value, acquire); / / - = 1 = 0-1-1
	if (likely(value >= 0)) {
		return 0;
	}
	return _dispatch_semaphore_wait_slow(dsema, timeout); / / go here
}
Copy the code

In the example above, the maximum number of concurrent semaphore_wait_slow is created with a semaphore of 0, which goes to wait, where 0-1 = -1 goes directly to _dispatch_semaphore_wait_slow

static intptr_t
_dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema,
		dispatch_time_t timeout) Parameter 1:0 parameter 2: FOREVER
{

	switch (timeout) {
	default:
		if(! _dispatch_sema4_timedwait(&dsema->dsema_sema, timeout)) {break;
		}
	// ...
	case DISPATCH_TIME_FOREVER:
		_dispatch_sema4_wait(&dsema->dsema_sema);
		break;
	}
	return 0;
}
Copy the code

void
_dispatch_sema4_wait(_dispatch_sema4_t *sema)
{
	int ret = 0;
	do {
		ret = sem_wait(sema);
	} while (ret == - 1 && errno == EINTR);
	DISPATCH_SEMAPHORE_VERIFY_RET(ret);
}
Copy the code

Sem_wait is the kernel code encapsulated by the underlying Pthread. We just need to pay attention to the do while loop here, which is essentially a do while loop waiting for the signal quantity to become positive.

Scheduling groupdispatch_group

Most direct: control the order in which tasks are executed

• dispatch_group_create: create a group
• dispatch_group_async: Group tasks
• dispatch_group_notify: indicates the completion of a group task
• dispatch_group_wait: indicates the waiting time for a group task to be executed
• dispatch_group_enter: into the group
• dispatch_group_leave: a group of

Scenario 1: Dispatch_group_async:

- (void)groupDemo{
    dispatch_group_t group = dispatch_group_create();
    dispatch_queue_t queue = dispatch_get_global_queue(0.0);
    
    dispatch_group_async(group, queue, ^{
    });
    
    dispatch_group_async(group, queue, ^{
        
    });
    dispatch_group_notify(group, dispatch_get_main_queue(), ^{

    });
    
}
Copy the code

Solution 2: Enter and leave

- (void)groupDemo{
    dispatch_group_t group = dispatch_group_create();
    dispatch_queue_t queue = dispatch_get_global_queue(0.0);
    
    dispatch_group_async(group, queue, ^{
    });
    
    dispatch_group_enter(group);
    dispatch_async(queue, ^{
       dispatch_group_leave(group);
    });
    
    dispatch_group_notify(group, dispatch_get_main_queue(), ^{

    });
}
Copy the code

The above two effects are the same. Why is “dispatch_group_ASYNC” = “dispatch_group_ENTER” + “dispatch_group_leave”

dispatch_group_create

dispatch_group_t
dispatch_group_create(void)
{
	return _dispatch_group_create_with_count(0);
}
Copy the code

The _dispatch_group_create_with_count function is similar to the semaphore function

dispatch_group_enter

I went into the source code and found that this is a — operation

dispatch_group_leave

void
dispatch_group_leave(dispatch_group_t dg)
{
	uint64_t new_state, old_state = os_atomic_add_orig2o(dg, dg_state,
			DISPATCH_GROUP_VALUE_INTERVAL, release); / / + + - 1 - > 0
	uint32_t old_value = (uint32_t)(old_state & DISPATCH_GROUP_VALUE_MASK);// -1 & maximum
		// old_value == DISPATCH_GROUP_VALUE_MASK
    	If old_value = -1, old_value = -1
	if (unlikely(old_value == DISPATCH_GROUP_VALUE_1)) {
		old_state += DISPATCH_GROUP_VALUE_INTERVAL;
		do {
			new_state = old_state;
			if ((old_state & DISPATCH_GROUP_VALUE_MASK) == 0) {
				new_state &= ~DISPATCH_GROUP_HAS_WAITERS;
				new_state &= ~DISPATCH_GROUP_HAS_NOTIFS;
			} else {
				new_state &= ~DISPATCH_GROUP_HAS_NOTIFS;
			}
			if (old_state == new_state) break;
		} while(unlikely(! os_atomic_cmpxchgv2o(dg, dg_state, old_state, new_state, &old_state, relaxed)));return _dispatch_group_wake(dg, old_state, true);
	}

	if (unlikely(old_value == 0)) {
		DISPATCH_CLIENT_CRASH((uintptr_t)old_value,
				"Unbalanced call to dispatch_group_leave()"); }}Copy the code

When dg = -1, it goes back to a do while loop until it wakes up _dispatch_group_wake, where dispatch_group_notify is awakened. Let’s go back to the analysis above. Let’s start with enter, where dg=-1, and wake up notify after the leave function exits the do while loop

dispatch_group_notify

static inline void
_dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
		dispatch_continuation_t dsn)
{
    / /...
	if ((uint32_t)old_state == 0) {
			os_atomic_rmw_loop_give_up({
					return _dispatch_group_wake(dg, new_state, false); }); }}Copy the code

Here you can see that when old_state == 0,_dispatch_group_wake starts the normal asynchronous or synchronous call out process of the block. If the notify is executed first, bind the block to the current group and wait until the notification of leaving the group is sent, _dispatch_group_wake(dg, old_state, true). This is why this method is called in two places. The main purpose of this method is to solve the timing problem of asynchronous loading.

dispatch_group_async = dispatch_group_enter + dispatch_group_leave

We know from the above example that these two are equivalent, so how does dispatch_group_async encapsulate the in-group and out-group implementations

void
dispatch_group_async(dispatch_group_t dg, dispatch_queue_t dq,
		dispatch_block_t db)
{
	dispatch_continuation_t dc = _dispatch_continuation_alloc();
	uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_GROUP_ASYNC;
	dispatch_qos_t qos;

	qos = _dispatch_continuation_init(dc, dq, db, 0, dc_flags);
	_dispatch_continuation_group_async(dg, dq, dc, qos);
}
Copy the code

static inline void
_dispatch_continuation_group_async(dispatch_group_t dg, dispatch_queue_t dq,
		dispatch_continuation_t dc, dispatch_qos_t qos)
{
	dispatch_group_enter(dg);
	dc->dc_data = dg;
	_dispatch_continuation_async(dq, dc, qos, dc->dc_flags);
}
Copy the code

And you can clearly see that there’s an operation called dispatch_continuation_group_async dispatch_group_Enter (dg), _dispatch_continuation_async -> dx_push -> _dispatch_root_queue_push -> _dispatch_root_queue_push_inline -> _dispatch_root_queue_poke -> _dispatch_root_queue_poke_slow -> _dispatch_root_queues_init -> _dispatch_root_queues_init_once -> _dispatch_worker_thread2 -> _dispatch_root_queue_drain -> _dispatch_continuation_POP_inline -> _dispatch_continuation_INVOke_inline -> _dispatch_continuation_INVOke_inline

		if (unlikely(dc_flags & DC_FLAG_GROUP_ASYNC)) {
			_dispatch_continuation_with_group_invoke(dc);
		} else {
			_dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
			_dispatch_trace_item_complete(dc);
		}
Copy the code

_dispatch_continuation_with_group_invoke

static inline void
_dispatch_continuation_with_group_invoke(dispatch_continuation_t dc)
{
	struct dispatch_object_s *dou = dc->dc_data;
	unsigned long type = dx_type(dou);
	if (type == DISPATCH_GROUP_TYPE) {
		_dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
		_dispatch_trace_item_complete(dc);
		dispatch_group_leave((dispatch_group_t)dou);
	} else {
		DISPATCH_INTERNAL_CRASH(dx_type(dou), "Unexpected object type"); }}Copy the code

After _dispatch_client_callout we see the outbound code dispatch_group_leave, so let’s think about it, the outbound group is not called until the task in the current queue has finished executing.

dispatch_source

GCD and runLoop are on the same level, and there is no so-called attribution relationship. Dispatch_source essentially controls the execution of blocks through conditions, and its CPU load is very small, so it does not occupy resources as much as possible. When one of its functions dispatch_source_merge_data is called from any thread, the handle (which can be simply interpreted as a block) defined by dispatch_source is executed. This process is called a Customevent user event.

• dispatch_source_create: create the source
• dispatch_source_set_event_handler: Sets the source event callback
• dispatch_source_merge_data : Source event Settings data
• dispatch_source_get_data: Obtains the source event data
• dispatch_resume: continue to
• dispatch_suspend: hang

The use method is relatively simple

- (void)demo {
    // 1. Create queues
    self.queue = dispatch_queue_create("hb.com".NULL);
    // 2. Create the source
    self.source = dispatch_source_create(DISPATCH_SOURCE_TYPE_DATA_ADD, 0.0, dispatch_get_main_queue());
    // 3. Source event callback
    dispatch_source_set_event_handler(self.source, ^{
        
        NSLog(@ "% @"[NSThread currentThread]);
        
        NSUInteger value = dispatch_source_get_data(self.source);
        self.totalComplete += value;
        NSLog(@" Progress: %.2f".self.totalComplete/100.0);
        self.progressView.progress = self.totalComplete/100.0;
    });
    
    self.isRunning = YES;
    dispatch_resume(self.source);
}
// 4. Dispatch_source_merge_data Modify the source data where used
/ / 5. Dispatch_resume continue
Copy the code

And this is not affected by runLoop is a workLoop, essentially a pthread underlayer encapsulation.

Supplement: Are mutable arrays thread safe

It is not safe to operate on the same array in multiple threads because simultaneous writes occur, which means it is not safe to operate on the same memory space at the same time. Whereas atomic can only secure itself and not external access, the solution is to add a barrier function that acts as a lock to mutable arrays