Shulou(Shulou.com)06/01 Report--

How to parse the ThreadPoolExecutor source code, in view of this problem, this article introduces the corresponding analysis and solution in detail, hoping to help more partners who want to solve this problem to find a more simple and feasible method.

I. Thread

Thread is the minimum operating unit of CPU scheduling. Thread model is divided into KLT model and ULT model, and JVM uses KLT model.

State of the thread: NEW,RUNNABLE,BLOCKED,TERMINATED

Second, thread pool 1. Two core issues solved by thread pools:

When performing a large number of asynchronous operations, the thread pool optimizes the system performance and reduces the system overhead caused by the repeated creation of threads.

Provides a way to limit and manage resources

2. 7 core parameters:

CorePoolSize: the number of core threads in the thread pool. When a task is submitted, the thread pool creates a new thread to execute the task until the current number of threads is equal to corePoolSize;. If the current number of threads is corePoolSize, the tasks that continue to be submitted are saved to the blocking queue, waiting to be executed. If the prestartAllCoreThreads () method of the thread pool is executed, the thread pool will create and start all core threads in advance.

MaximumPoolSize: the maximum number of threads allowed in the thread pool. If the current blocking queue is full and the task continues to be submitted, a new thread is created to execute the task, provided that the current number of threads is less than maximumPoolSize

KeepAliveTime: the thread pool maintains the idle time allowed by threads. When the number of threads in the thread pool is large corePoolSize, if no new task is submitted at this time, threads outside the core thread will not be destroyed immediately, but will wait until the waiting time exceeds the keepAliveTime.

Unit: the unit of keepAliveTime

WorkQueue: used to hold the blocking queue for tasks waiting to be executed, and the task must implement the Runable interface. The following blocking queues are provided in JDK:

ArrayBlockingQueue: a bounded blocking queue based on array structure, sorting tasks by FIFO

LinkedBlockingQuene: a blocking queue based on a linked list structure that sorts tasks by FIFO, and the throughput is usually higher than that of ArrayBlockingQuene;-

SynchronousQuene: a blocking queue that does not store elements. Each insert operation must wait until another thread invokes the remove operation, otherwise the insert operation is always blocked and the throughput is usually higher than LinkedBlockingQuene.

PriorityBlockingQuene: unbounded blocking queue with priority

ThreadFactory: this is a variable of type ThreadFactory that is used to create a new thread. Executors.defaultThreadFactory () is used by default to create threads. When you use the default ThreadFactory to create a thread, the newly created thread has the same NORM_PRIORITY priority and is not a daemon thread, and the name of the thread is also set.

Handler: thread pool saturation strategy. When the blocking queue is full and there are no idle worker threads, if you continue to submit a task, you must adopt a strategy to deal with the task. Thread pool provides four strategies:

AbortPolicy: throw an exception directly. Default policy

CallerRunsPolicy: use the caller's thread to execute the task

DiscardOldestPolicy: discards the first task in the blocking queue and executes the current task

DiscardPolicy: discarding tasks directly

The above four strategies are all inner classes of ThreadPoolExecutor. Of course, you can also implement the RejectedExecutionHandler interface according to the application scenario and customize the saturation policy, such as logging or tasks that cannot be handled by persistent storage.

3. Lifecycle state of the thread pool:

NEW

RUNNABLE

WATING

BLOCKED

TIMED_WATING

TERMINATED

4. Important attribute of thread pool: ctl

Ctl is a field that controls the running status of the thread pool and the number of valid threads in the thread pool. It contains two parts of information: the running status of the thread pool (runState) and the number of valid threads in the thread pool (workerCount). It can be seen that the Integer type is used to save, high 3 bits save runState, low 29 bits save workerCount. COUNT_BITS is 29 catechy, which means 1 shift to the left 29 bits minus 1 (29 ones). This constant represents the upper limit of workerCount, which is about 500 million.

RunState mainly provides control of the thread pool life cycle, and the main values include:

RUNNING

(1) status description: when the thread pool is in the RUNNING state, it can receive new tasks and process the added tasks.

(2) State switching: the initialization state of the thread pool is RUNNING. In other words, once the thread pool is created, it is in the RUNNING state, and the number of tasks in the thread pool is 0!

SHUTDOWN

(1) status description: when the thread pool is in the SHUTDOWN state, it does not receive new tasks, but can handle added tasks.

(2) State switching: when calling the shutdown () interface of the thread pool, the thread pool is RUNNING->-SHUTDOWN.

STOP

(1) status description: when the thread pool is in the STOP state, it does not receive new tasks, does not process the added tasks, and will break the tasks being processed.

(2) State switching: when calling the shutdownNow () API of the thread pool, the thread pool consists of (RUNNING or SHUTDOWN)-> STOP.

TIDYING

(1) status description: when all tasks have been terminated, the "number of tasks" recorded by ctl is 0, and the thread pool will become TIDYING. When the thread pool changes to the TIDYING state, the hook function terminated () is executed. Terminated () is empty in the ThreadPoolExecutor class, and if the user wants to handle it when the thread pool becomes TIDYING, it can be done by overloading the terminated () function.

(2) State switching: when the thread pool is in the SHUTDOWN state, the blocking queue is empty and the tasks executed in the thread pool are also empty, it will be SHUTDOWN-> TIDYING. When the thread pool is in the STOP state and the task executed in the thread pool is empty, it will be STOP-> TIDYING.

TERMINATED

(1) status description: when the thread pool terminates completely, it becomes TERMINATED state.

(2) State switching: when the thread pool is in TIDYING state, after executing terminated (), it will be TIDYING-> TERMINATED. The conditions for entering TERMINATED are as follows: thread pool is not in RUNNING state; thread pool state is not TIDYING state or TERMINATED state; if thread pool state is SHUTDOWN and workerQueue is empty; if workerCount is 0, the TIDYING state is set successfully.

Ctl related API

RunStateOf (): get the running status

WorkerCountOf (): get number of active threads

CtlOf (): gets the values of running status and the number of active threads.

5. Behavior of thread pools

Execute (Runnable command): perform tasks of type Ruannable

Submit (task): can be used to submit a Callable or Runnable task and return a Future object that represents this task

Shutdown (): close the work after completing the submitted task and no longer take over the new task

ShutdownNow (): stop all tasks that are being performed and close the work.

IsTerminated (): test whether all tasks have been performed.

IsShutdown (): test whether the ExecutorService has been closed.

6. The specific implementation of common thread pool ThreadPoolExecutor default thread pool ScheduledThreadPoolExecutor timing thread pool 7. Thread pool monitoring API

Public long getTaskCount () / / Total number of executed and unexecuted tasks in the thread pool

Public long getCompletedTaskCount () / / number of tasks completed

Current number of threads in the public int getPoolSize () / / thread pool

Public int getActiveCount () / / the number of threads executing tasks in the thread pool

Third, the source code parses the execute () method / / to perform a given task at some point in the future. The task can be a new thread or reuse threads in the existing pool to execute public void execute (Runnable command) {if (command = = null) throw new NullPointerException (); / * the execution process is still divided into three steps: * * 1. Execute tasks: * if it is less than the number of core threads, try to create a new thread to perform the given task. * the method addWorker () is the way to actually create a new thread to execute the task. * the addWorker () method performs atomic checks on runState and workerCount. * the addWorker () method returns a boolean value, which prevents a false alarm * 2 if a thread should not be added by returning a false value. Add to the blocking queue: * if you fail to meet the conditions and execute step 1, add to the blocking queue. * if the task can be queued successfully, it will check again to see if a thread should be added (because the existing thread has died since the last check), or the pool has been closed since entering this method. Therefore, you need to re-check the status and roll back the queue if necessary if stopped, and if not, start a new thread. * 3. Reject task: * if the task cannot be added to the blocking queue, the maximum number of threads does not reach the maximum to try to add a new thread. If it fails, the thread pool is closed or saturated, so the task is rejected. * / / clt records runState and workerCount int c = ctl.get (); / * * the workerCountOf method takes a value as low as 29 bits to indicate the number of currently active threads; * if the number of currently active threads is less than corePoolSize, create a new thread into the thread pool and add tasks to the thread. * / if (workerCountOf (c)

< corePoolSize) { /* * addWorker中的第二个参数表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断 * 如果为true，根据corePoolSize来判断; * 如果为false，则根据maximumPoolSize来判断 */ if (addWorker(command, true)) return; //如果添加失败，则重新获取ctl值 c = ctl.get(); } //执行到此处说明从核心线程里给当前任务分配线程失败 //如果当前线程池是运行状态并且任务添加到队列成功 if (isRunning(c) && workQueue.offer(command)) { //重新获取ctl值。即使添加队列成功也要再次检查，如果不是运行状态，由于之前已经把任务添加到workerQueue 中了，所以要移除该任务,执行过后通过handler使用拒绝策略对该任务进行处理，整个方法返回 int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); /* * 获取线程池中的有效线程数，如果数量是0，则执行addWorker方法 这里传入的参数表示: * 第一个参数为null，表示在线程池中创建一个线程，但不去启动; * 第二个参数为false，将线程池的有限线程数量的上限设置为maximumPolSize，添加线程时根据maximumPoolSize来判断; * 如果判断workerCount大于0，则直接返回，在workQueue中新增的comman 会在将来的某个时刻被执行。 */ //因为 任务已经被添加到workQueue中了，所以worker在执行的时候，会直接从workQueue中 获取任务。 else if (workerCountOf(recheck) == 0) //执行到这里说明任务已经添加到阻塞队列里了，最大线程数也未饱和，则创建一个新的线程去阻塞队列里拿任务 //这步操作也就是创建一个线程，但并没有传入任务，因为任务已经被添加到workQueue中了，所以worker在执行的时候，会直接从workQueue中 获取任务。 //为什么要这样做呢？是为了保证线程池在RUNNING状态下必须要有一个线程来执行任务。 addWorker(null, false); } /* * 如果执行到这里，有两种情况: * 1. 线程池已经不是RUNNING状态; * 2. 线程池是RUNNING状态，但workerCount >

= corePoolSize and workQueue is full. * at this time, the addWorker method is called again, but the second parameter is passed to false, and the limit of the limited number of threads in the thread pool is set to maximumPoolSize; * reject the task if it fails * / else if (! addWorker (command, false)) reject (command) The} addWorker () method / * checks whether a new worker thread can be added to execute the task based on the state of the current thread pool. * addWorker (runnable,true) means that threads are assigned from the number of core worker threads to perform incoming tasks; * addWorker (null,false) means that threads are allocated from the maximum number of threads to perform tasks in the blocking queue. * if the thread pool is stopped or closed, it will directly return false. If the thread pool fails to create a new thread, it will also return false. * if the creation of a thread fails, or the thread factory returns null, or the thread executing the current addWorker () throws an exception, (note that the current thread throws an exception, which is only related to the current task and does not affect the execution of other tasks) The relevant properties of the thread pool will be rolled back immediately * / private boolean addWorker (Runnable firstTask, boolean core) {retry: / / the outer for loop is to be ready to assign threads to the task, to judge the state-- > atomic increment workerCount / / until the thread pool state does not meet the criteria and returns false, or the self-increment successfully jumps out of the for loop / / the same GetTask () has the same logic when getting a task from a blocking queue, first decrementing the workerCount atom, and then executing the task for ( ;) {/ / you can see that each step determines the state parameters of the thread pool int c = ctl.get (); int rs = runStateOf (c) / / also check the thread pool status and queue status / * the state judgment here is also easy to understand: * Thread pool status is SHUTDOWN, no new tasks will be accepted, return false * if the city status is not SHUTDOWN, the incoming task is empty, and there are no tasks in the blocking queue, then perform a hammer task. Also return false * / if (rs > = SHUTDOWN & &! (rs = = SHUTDOWN & & firstTask = = null &! WorkQueue.isEmpty ()) return false; / / has previously judged that it is satisfied to assign a thread to the task to execute the task / / this for loop is to prepare for the creation of the task. First, the atomic incremental workerCount,workerCount increment is successful before the real thread is assigned to the task to execute for ( ) {/ / the current number of worker threads int wc = workerCountOf (c); / / the current number of worker threads is greater than corePoolSize or maximumPoolSize (the comparison is based on the parameter core passed in), and / / indicates that no assigned thread can execute the task. Return false if (wc > = CAPACITY | | wc > = (core? CorePoolSize: maximumPoolSize)) return false / * the execution here indicates that the conditions are met and threads can be assigned to execute the task * try to add workerCount, and if successful, jump out of the first for loop * here is the workerCount for CAS self-increment ctl (increase the number first Then jump out of the for loop to create a new thread to execute the task) * the atomic class AtomicInteger.compareAndSet () method is also called inside this method to ensure that the atomic increment * / if (compareAndIncrementWorkerCount (c)) break retry / / if the attempt to add a new worker thread fails, it will continue to judge the status of the current thread pool and continue to try to assign the worker thread to the current thread c = ctl.get (); / / Re-read ctl if (runStateOf (c)! = rs) continue retry; / / else CAS failed due to workerCount change Retry inner loop}} / / after jumping out of the for cycle, the number of worker threads in the thread pool has been adjusted by workerCount. The next step is to actually assign threads and execute the task boolean workerStarted = false; boolean workerAdded = false; Worker w = null. The try {/ / Worker object is an inner class. In fact, it uses threatFactory to generate a new thread / / inherit AQS class, implement the Runable interface, override the run () method, and the rewritten run () method is also important. W = new Worker (firstTask); final Thread t = w.thread If (t! = null) {/ / lock guarantee synchronization final ReentrantLock mainLock = this.mainLock; mainLock.lock (); try {/ / first do a thread pool status check int rs = runStateOf (ctl.get ()); if (rs)

< SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //这个 workers 是个 HashSet，线程池也是通过维护这个 workers 控制任务的执行 workers.add(w); int s = workers.size(); if (s >

LargestPoolSize) / / largestPoolSize records the maximum number of threads that have appeared in the thread pool largestPoolSize = s; workerAdded = true;}} finally {mainLock.unlock () } if (workerAdded) {/ / finally, call the thread's start () method t.start (); workerStarted = true;}} finally {/ / if thread creation fails, roll back the state of the thread pool if (! WorkerStarted) addWorkerFailed (w);} return workerStarted;} Worker class

What is the Worker class for? the meaning of existence.

Looking back at the Worker class, each thread in the thread pool is encapsulated into a Worker object, and ThreadPool maintains a set of Worker objects (HashSet).

The Worker class inherits AQS and implements the Runnable interface. Note the attributes of firstTask and thread: firstTask uses it to hold incoming tasks; thread is a thread created through ThreadFactory when a constructor is called and is used to process tasks.

Worker inherits AQS and uses AQS to implement the function of exclusive lock. Why not use ReentrantLock to implement it? You can see the tryAcquire method, which does not allow reentry, while ReentrantLock allows reentry:

Once the lock method acquires an exclusive lock, it indicates that the current thread is executing a task

If you are performing a task, you should not interrupt the thread

If the thread is not in an exclusive lock state, that is, an idle state, it is not processing a task, and the thread can be interrupted.

When the thread pool executes the shutdown method or the tryTerminate method, the interruptIdleWorkers method is called to interrupt the idle thread. The interruptIdleWorkers method uses the tryLock method to determine whether the thread in the thread pool is idle.

It is set to non-reentrant because we do not want the task to reacquire the lock when calling thread pool control methods such as setCorePoolSize. If you use ReentrantLock, it is reentrant, so that if a thread pool control method such as setCorePoolSize is called in a task, the running thread will be interrupted. Therefore, Worker inherits from AQS and is used to determine whether a thread is idle and can be interrupted. In addition, setState (- 1); is executed in the constructor, and the state variable is set to-1. Why would you do that? Because the default state in AQS is 0, if a Worker object has just been created and the task has not been executed, it should not be interrupted at this time. Take a look at the tryAquire method:

Important methods in and involved in the Worker class

TryAcquire (int unused) method

/ * used to determine whether the thread is idle and whether it can be interrupted * / protected boolean tryAcquire (int unused) {/ / cas modify the state, not reenter if (compareAndSetState (0,1)) {setExclusiveOwnerThread (Thread.currentThread ()); return true;} return false;}

The tryAcquire method is based on whether the state is 0 or not, so setting state to-1 is to prevent interruptions to the thread before the task is executed.

Because of this, the unlock method of the Worker object is first called in the runWorker method to set the state to 0.

RunWorker (Worker w) method

/ * the Worker class implements the Runnable interface, and the overridden run () method * / final void runWorker (Worker w) {Thread wt = Thread.currentThread (); Runnable task = w.firstTask; w.firstTask = null; / / allows interrupts w.unlock (); / / allow interrupts boolean completedAbruptly = true The try {/ / while loop is to execute the task continuously, and still take the task from the blocking queue after the execution of its own task (firstTask). This operation ensures that the thread's reuse / / task is empty, then get the task while from the blocking queue (task! = null | | (task = getTask ())! = null) {w.lock () / / If pool is stopping, ensure thread is interrupted; / / if not, ensure thread is not interrupted. This / / requires a recheck in second case to deal with / / shutdownNow race while clearing interrupt / * if the thread pool is stopping, make sure that the current thread is in the interrupted state, and if not, make sure the current thread is not in the interrupted state * Why are you doing this here? Considering that the shutdownNow method may also have been executed during the execution of the if statement, the shutdownNow method sets the state to STOP, * Review the STOP status: you cannot accept new tasks, do not process tasks in the queue, and interrupt threads that are processing tasks. When the thread pool is in the RUNNING or SHUTDOWN state, * calling the shutdownNow () method puts the thread pool into the STOP state. * the STOP state interrupts all threads in the thread pool. Here, Thread.interrupted () is used to determine whether the thread is in a non-interrupted state in the RUNNING or SHUTDOWN state, * because the Thread.interrupted () method resets the interrupted state. * / if ((runStateAtLeast (ctl.get (), STOP) | | (Thread.interrupted () & & runStateAtLeast (ctl.get (), STOP) & & wt.isInterrupted () wt.interrupt (); try {beforeExecute (wt, task); Throwable thrown = null Try {task.run ();} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; throw new Error (x) } finally {afterExecute (task, thrown);}} finally {task = null; W.completedTasksgiving examples; w.unlock ();}} completedAbruptly = false;} finally {processWorkerExit (w, completedAbruptly);}}

Summarize the execution of the runWorker method:

The while loop constantly acquires tasks through the getTask () method

The getTask () method fetches tasks from the blocking queue

If the thread pool is stopping, make sure that the current thread is in the interrupted state, otherwise the current thread is not in the interrupted state.

Call task.run () to execute the task

If task is null, jump out of the loop and execute the processWorkerExit () method

The completion of the execution of the runWorker method also means that the execution of the run method in the Worker is complete and the thread is destroyed.

GetTask () method

/ * get a task from the blocking queue. The return value is Runnable * if the thread pool state does not meet the execution condition, the value of the variable null / private Runnable getTask () {/ / timeOut indicates whether the last time the task was fetched from the blocking queue timed out boolean timedOut = false / / Did the last poll () time out? / / the two for loop operations here are the same as the two for loop operations in the addWorker () method. For (;) {int c = ctl.get (); int rs = runStateOf (c); / / still check the thread pool state / / Check if queue empty only if necessary. If (rs > = SHUTDOWN & & (rs > = STOP | | workQueue.isEmpty ()) {decrementWorkerCount (); return null;} int wc = workerCountOf (c) The / / timed variable is used to determine whether timeout control is required / / allowCoreThreadTimeOut is false by default, that is, core threads are not allowed to timeout / / wc > corePoolSize, indicating that the number of threads in the current thread pool is greater than the number of core threads / / for these threads that exceed the number of core threads, timeout control / / Are workers subject to culling is required? Boolean timed = allowCoreThreadTimeOut | | wc > corePoolSize / / as with the process in addWorker (), the workerCount in the thread pool is controlled first, and then the subsequent task operations / / if the conditions are met, the number of workerCount is reduced by one if ((wc > maximumPoolSize | | (timed & & timedOut)) & & (wc > 1 | | workQueue.isEmpty () {if (compareAndDecrementWorkerCount (c) return null; continue) } try {/ / according to timed, if it is true, the timeout is controlled by blocking the poll method of the queue. If the task is not obtained within the keepAliveTime time, null / / is returned. Otherwise, through the take method, if the queue is empty, the take method blocks until the queue is not empty Runnable r = timed? WorkQueue.poll (keepAliveTime, TimeUnit.NANOSECONDS): workQueue.take (); if (r! = null) return r; / / if r = = null, the timeout has occurred, and timedOut is set to true timedOut = true } catch (InterruptedException retry) {/ / if the current thread is interrupted when getting the task, set timedOut to false and return timedOut = false;}} processWorkerExit () / * getTask method returns null, the while loop will jump out of the runWorker method, and then the processWorkerExit method will be executed. * do the clean-up work of the thread pool * / private void processWorkerExit (Worker w, boolean completedAbruptly) {/ / if the completedAbruptlyvalue is true, there is an exception during thread execution, and you need to subtract workerCount by 1 / / if there is no exception during thread execution, which means that workerCount has already been subtracted by 1 in the getTask () method, so there is no need to subtract if (completedAbruptly) / / If abrupt, then workerCount wasn't adjusted decrementWorkerCount () Final ReentrantLock mainLock = this.mainLock; mainLock.lock (); try {/ / counts the number of tasks completed completedTaskCount + = w.completedTasks; / / removed from the workers, which means that a worker thread is removed from the thread pool / / workers is the aforementioned HashSet, and the thread pool is the workers.remove (w) that maintains this worker () to keep the thread pool running. } finally {mainLock.unlock ();} / / decide whether to end the thread pool tryTerminate () according to the thread pool status; / * if the thread pool is in RUNNING or SHUTDOWN state, if the worker ends abnormally, it will directly addWorker. * if allowCoreThreadTimeOut=true and there is a task in the waiting queue, keep at least one worker * if allowCoreThreadTimeOut=false,workerCount is not less than corePoolSize * / int c = ctl.get (); if (runStateLessThan (c, STOP)) {if (! completedAbruptly) {int min = allowCoreThreadTimeOut? 0: corePoolSize; if (min = 0 & &! WorkQueue.isEmpty () min = 1; if (workerCountOf (c) > = min) return; / / replacement not needed} addWorker (null, false);}}

At this point, after the execution of processWorkerExit, the worker thread is destroyed. This is the life cycle of the entire worker thread. Starting from the execute method, Worker uses ThreadFactory to create a new worker thread, runWorker gets the task through getTask, and then executes the task. If getTask returns null, enter the processWorkerExit method, and the whole thread ends.

IV. Thinking

How does thread pool achieve thread reuse?

It is in the overridden run () method that through the while loop, the task is still fetched from the blocking queue to execute after the firstTask is executed.

How to deal with thread timeout?

When the previous task throws an exception, will the subsequent thread still execute? The answer is yes. It is also found in the while loop. The exception thrown by the current thread will only affect the current thread, not other tasks in the thread pool.

When will it be destroyed?

After the execution of the runWorker method, that is, the execution of the run method in Worker, it is automatically reclaimed by JVM.

Blocking queue selection? The following blocking queues are provided in JDK:

ArrayBlockingQueue: a bounded blocking queue based on array structure, sorting tasks by FIFO

LinkedBlockingQuene: a blocking queue based on linked list structure. Tasks are sorted by FIFO, and the throughput is usually higher than ArrayBlockingQuene.

SynchronousQuene: a blocking queue that does not store elements. Each insert operation must wait until another thread invokes the remove operation, otherwise the insert operation is always blocked and the throughput is usually higher than LinkedBlockingQuene.

PriorityBlockingQuene: unbounded blocking queue with priority

Discard strategy selection? Thread pools provide four strategies:

AbortPolicy: throw an exception directly. Default policy

CallerRunsPolicy: use the caller's thread to execute the task

DiscardOldestPolicy: discards the first task in the blocking queue and executes the current task

DiscardPolicy: discarding tasks directly

How to set the number of threads?

The general idea is to set it according to the type of our task, which is simply divided into: CPU-intensive: CPU core + 1 IO-intensive: 2*CPU core + 1

The original formula in Java concurrent programming practice is as follows: Nthreads=Ncpu ∗ Ucpu ∗ (1+WC) Nthreads=Ncpu ∗ Ucpu ∗ (1+CW)

Ncpu represents the number of CPU

Ucpu represents the expected value of CPU utilization (0

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.